from . import testutils

if __name__ == "__main__":

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

import pkg5unittest

import os

import pkg.config as cfg

import pkg.fmri as fmri

import pkg.misc as misc

import pkg.server.repository as repo

import shutil

import tempfile

import time

import unittest

import zlib

import sys

class TestUtilMerge(pkg5unittest.ManyDepotTestCase):

persistent_setup = True

scheme10 = """

tree10 = """

amber10 = """

amber20 = """

bronze10 = """

bronze20 = """

bronze20b = """

bronze20c = """

silverA = """

silverB = """

# etc/tree is purposefully delivered by this package to verify that

# pkgmerge can have an identical file being present in multiple

# packages (this and scheme@1.0) with different publishers.

multiA = """

multiB = """

multiC = """

multiD = """

tinA = """

tinB = """

mediatorPPC = """

mediatorARM = """

misc_files = [ "tmp/bronzeA1", "tmp/bronzeA2", "tmp/bronze1",

"tmp/bronze2", "tmp/copyright2", "tmp/copyright3", "tmp/libc.so.1",

"tmp/sh", "tmp/scheme", "tmp/sparc-only", "tmp/sparc1", "tmp/sparc2",

"tmp/sparc3", "tmp/sparc4", "tmp/i3861", "tmp/i3862", "tmp/i3863"]

def setUp(self):

pkg5unittest.ManyDepotTestCase.setUp(self, 18 * ["os.org"])

self.make_misc_files(self.misc_files)

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

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

self.rurl3 = self.dcs[3].get_repo_url()

self.rurl4 = self.dcs[4].get_repo_url()

self.rurl5 = self.dcs[5].get_repo_url()

self.rurl6 = self.dcs[6].get_repo_url()

# Empty repository.

self.rurl7 = self.dcs[7].get_repo_url()

# a bunch for testing blending

self.rurl8 = self.dcs[8].get_repo_url()

self.rurl9 = self.dcs[9].get_repo_url()

self.rurl10 = self.dcs[10].get_repo_url()

self.rurl11 = self.dcs[11].get_repo_url()

self.rurl12 = self.dcs[12].get_repo_url()

self.rurl13 = self.dcs[13].get_repo_url()

# repositories which will contain several publishers

self.rurl14 = self.dcs[14].get_repo_url()

self.rurl15 = self.dcs[15].get_repo_url()

# mediator testing

self.rurl16 = self.dcs[16].get_repo_url()

self.rurl17 = self.dcs[17].get_repo_url()

# Publish a set of packages to one repository.

self.published = self.pkgsend_bulk(self.rurl1, (self.amber10,

self.amber20, self.bronze10, self.bronze20, self.tree10,

self.scheme10))

# Ensure timestamps of all successive publications are greater.

time.sleep(1)

# Publish the same set to another repository (minus the tree

# and scheme packages, and with a slightly different version

# of the bronze20 package).

self.published += self.pkgsend_bulk(self.rurl2, (self.amber10,

self.amber20, self.bronze10, self.bronze20b))

# Ensure timestamps of all successive publications are greater.

time.sleep(1)

# Publish the same set to another repository (with a slightly

# different version of the bronze20b package and with tree).

self.published += self.pkgsend_bulk(self.rurl3, (self.amber10,

self.amber20, self.bronze10, self.bronze20c, self.tree10))

# Ensure timestamps of all successive publications are greater.

time.sleep(1)

# Everything above again, but this time using the debug version

# of the misc. files and a different set of repositories.

dfiles = dict((f, f + ".debug") for f in self.misc_files)

# For testing purposes, don't change this file for the debug

# variant case.

dfiles["tmp/libc.so.1"] = "tmp/libc.so.1"

self.make_misc_files(dfiles)

self.published_debug = self.pkgsend_bulk(self.rurl4,

(self.amber10, self.amber20, self.bronze10, self.bronze20,

self.tree10, self.scheme10))

time.sleep(1)

self.published_debug += self.pkgsend_bulk(self.rurl5,

(self.amber10, self.amber20, self.bronze10, self.bronze20b))

time.sleep(1)

self.published_debug += self.pkgsend_bulk(self.rurl6, (

self.amber10, self.amber20, self.bronze10, self.bronze20c,

self.tree10))

self.published_blend = self.pkgsend_bulk(self.rurl8, (self.silverA,

self.tinA))

time.sleep(1)

self.published_blend += self.pkgsend_bulk(self.rurl9, (self.silverB,

self.tinB))

time.sleep(1)

self.published_blend += self.pkgsend_bulk(self.rurl10, (self.multiA,))

time.sleep(1)

self.published_blend += self.pkgsend_bulk(self.rurl11, (self.multiB,))

time.sleep(1)

self.published_blend += self.pkgsend_bulk(self.rurl12, (self.multiC,))

time.sleep(1)

self.published_blend += self.pkgsend_bulk(self.rurl13, (self.multiD,))

# Publish to multiple repositories, maintaining lists of which

# FMRIs are published to which repository.

self.published_multi_14 = []

self.published_multi_15 = []

for url, record in [

(self.rurl14, self.published_multi_14),

(self.rurl15, self.published_multi_15)]:

time.sleep(1)

record += self.pkgsend_bulk(url, (self.scheme10))

time.sleep(1)

record += self.pkgsend_bulk(url, (self.tree10))

time.sleep(1)

record += self.pkgsend_bulk(url,

self.bronze20.replace("open ",

"open pkg://altpub/"))

time.sleep(1)

record += self.pkgsend_bulk(url,

(self.amber10.replace("open ",

"open pkg://altpub/")))

time.sleep(1)

record += self.pkgsend_bulk(url,

(self.multiA.replace("open ",

"open pkg://last/")))

# add bronze20b to one repository so that we have at least one

# package where more complex merging happens.

time.sleep(1)

self.published_multi_15 += self.pkgsend_bulk(self.rurl15,

self.bronze20b.replace("open ", "open pkg://altpub/"))

# one of our source repositories also contains a newer

# version of pkg:/gold (self.multi*)

time.sleep(1)

self.published_multi_15 += self.pkgsend_bulk(self.rurl15,

(self.multiB.replace("open ", "open pkg://last/")))

# publish to our mediator repos

self.published_16 = self.pkgsend_bulk(self.rurl16,

(self.mediatorPPC))

self.published_17 = self.pkgsend_bulk(self.rurl17,

(self.mediatorARM))

def test_0_options(self):

"""Verify that pkgmerge gracefully fails when given bad option

# Should fail because no source was specified.

self.pkgmerge(" ".join([

"-d {0}".format(self.rurl2),

]), exit=2)

# Should fail because no destination was specified.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl2),

]), exit=2)

# Should fail because variant for source was not provided.

self.pkgmerge(" ".join([

"-s {0}".format(self.rurl1),

"-d {0}".format(self.rurl2),

]), exit=2)

# Should fail because only source variant was provided.

self.pkgmerge(" ".join([

"-s arch=i386",

"-d {0}".format(self.rurl2),

]), exit=2)

# Should fail because user did not specify the same variants

# for every source.

self.pkgmerge(" ".join([

"-s arch=sparc,{0}".format(self.dcs[1].get_repodir()),

"-s arch=i386,debug=true,{0}".format(self.dcs[2].get_repodir()),

"-d {0}".format(self.rurl7)

]), exit=2)

# Should fail because user did not specify a source for all

# variant combinations (e.g. i386 & arm debug).

self.pkgmerge(" ".join([

"-s arch=sparc,debug=false,{0}".format(self.dcs[1].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[2].get_repodir()),

"-s arch=arm,debug=false,{0}".format(self.dcs[3].get_repodir()),

"-s arch=sparc,debug=true,{0}".format(self.dcs[4].get_repodir()),

"-d {0}".format(self.rurl7)

]), exit=2)

# Should fail because source is not a repository.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.test_root),

"-d {0}".format(self.rurl2),

]), exit=1)

# Should fail because destination is not a repository.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl2),

"-d {0}".format(self.test_root),

]), exit=1)

# Should fail because of no matching -p publishers.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl2),

"-d {0} -p noodles".format(self.test_root),

]), exit=1)

# Create the target repository.

repodir = os.path.join(self.test_root, "test_0_repo")

self.create_repo(repodir)

# Should fail because of no matching packages.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl1),

"-d {0} nomatching".format(repodir),

]), exit=1)

shutil.rmtree(repodir)

def test_1_single_merge(self):

"""Verify that merge functionality works as expected when

#

# First, verify that merging all packages from a single source

# works as expected.

#

# Create the target repository.

repodir = os.path.join(self.test_root, "1merge_repo")

self.create_repo(repodir)

# Perform a dry-run.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl2),

"-d {0}".format(repodir),

"-n",

]))

# Merge the packages.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl2),

"-d {0}".format(repodir),

]))

# Get target repository catalog.

repo = self.get_repo(repodir)

cat = repo.get_catalog(pub="os.org")

# Verify the list of expected packages in the target repository.

expected = [

self.published[7], # pkg://os.org/amber@2.0-0

self.published[9], # pkg://os.org/bronze@2.0-0

]

actual = [str(f) for f in sorted(cat.fmris())]

self.assertEqualDiff(expected, actual)

# Verify that each package was merged correctly.

merged_expected = {

"amber": """\

""".format(self.published[7]), # pkg://os.org/amber@2.0-0

"bronze": """\

""".format(self.published[9]) # pkg://os.org/bronze@2.0-0

}

for f in cat.fmris():

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

expected = merged_expected[f.pkg_name]

self.assertEqualDiff(expected, actual)

#

# Next, verify that merging specific packages from a single

# source works as expected.

#

# Create the target repository.

shutil.rmtree(repodir)

self.create_repo(repodir)

# Perform a dry-run.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl2),

"-d {0}".format(repodir),

"-n",

"amber@latest",

]))

# Merge the packages.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl2),

"-d {0}".format(repodir),

"amber@latest",

]))

# Get target repository catalog.

repo = self.get_repo(repodir)

cat = repo.get_catalog(pub="os.org")

# Verify the list of expected packages in the target repository.

expected = [

self.published[7], # pkg://os.org/amber@2.0-0

]

actual = [str(f) for f in sorted(cat.fmris())]

self.assertEqualDiff(expected, actual)

# Verify that each package was merged correctly.

for f in cat.fmris():

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

expected = merged_expected[f.pkg_name]

self.assertEqualDiff(expected, actual)

# Next, verify that an attempt to merge using a variant that

# doesn't match what is already declared in a source's packages

# results in failure. (e.g. packages are tagged i386, but

# source was claimed to be sparc.)

junk_repodir = os.path.join(self.test_root, "1junk_repo")

self.create_repo(junk_repodir)

self.pkgmerge(" ".join([

"-s arch=sparc,{0}".format(repodir),

"-d {0}".format(junk_repodir),

"amber@latest",

]), exit=1)

# Cleanup.

shutil.rmtree(repodir)

def test_2_multi_merge(self):

"""Verify that merge functionality works as expected when

#

# First, verify that merging all packages from multiple sources

# works as expected when specifying two variant values.

#

# Create the target repository.

repodir = os.path.join(self.test_root, "2merge_repo")

self.create_repo(repodir)

# Merge the packages.

self.pkgmerge(" ".join([

"-s arch=i386,{0}".format(self.rurl2),

"-s arch=sparc,{0}".format(self.rurl1),

"-d {0}".format(repodir)

]))

# Get target repository catalog.

repo = self.get_repo(repodir)

cat = repo.get_catalog(pub="os.org")

# Verify the list of expected packages in the target repository.

nlist = {}

for s in self.published[:10]:

f = fmri.PkgFmri(s)

if f.pkg_name not in nlist or \

f.version > nlist[f.pkg_name].version:

nlist[f.pkg_name] = f

nlist = sorted(nlist.values())

expected = [str(f) for f in nlist]

actual = [str(f) for f in sorted(cat.fmris())]

self.assertEqualDiff(expected, actual)

# Verify that each package was merged correctly.

merged_expected = {

"amber": """\

""".format(self.published[7]), # pkg://os.org/amber@2.0-0

"bronze": """\

""".format(self.published[9]), # pkg://os.org/bronze@2.0-0

"scheme": """\

""".format(self.published[5]), # pkg://os.org/scheme@1.0-0

"tree": """\

""".format(self.published[4]), # pkg://os.org/tree@1.0-0

}

for f in nlist:

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

expected = merged_expected[f.pkg_name]

self.assertEqualDiff(expected, actual)

#

# Next, verify that merging packages for three variant values

# works as expected.

#

# Create the target repository.

shutil.rmtree(repodir)

self.create_repo(repodir)

# Merge the packages.

self.pkgmerge(" ".join([

"-s arch=sparc,{0}".format(self.rurl1),

"-s arch=i386,{0}".format(self.rurl2),

"-s arch=arm,{0}".format(self.rurl3),

"-d {0}".format(repodir)

]))

# Get target repository catalog.

repo = self.get_repo(repodir)

cat = repo.get_catalog(pub="os.org")

# Verify the list of expected packages in the target repository.

nlist = {}

for s in self.published:

f = fmri.PkgFmri(s)

if f.pkg_name not in nlist or \

f.version > nlist[f.pkg_name].version:

nlist[f.pkg_name] = f

nlist = sorted(nlist.values())

expected = [str(f) for f in nlist]

actual = [str(f) for f in sorted(cat.fmris())]

self.assertEqualDiff(expected, actual)

# Verify that each package was merged correctly.

merged_expected = {

"amber": """\

""".format(self.published[11]), # pkg://os.org/amber@2.0-0

"bronze": """\

""".format(self.published[13]), # pkg://os.org/bronze@2.0-0

"scheme": """\

""".format(self.published[5]), # pkg://os.org/scheme@1.0-0

"tree": """\

""".format(self.published[14]), # pkg://os.org/tree@1.0-0

}

for f in nlist:

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

expected = merged_expected[f.pkg_name]

self.assertEqualDiff(expected, actual)

#

# Next, verify that merging specific packages from multiple

# sources works as expected.

#

# Create the target repository.

shutil.rmtree(repodir)

self.create_repo(repodir)

# Merge the packages.

self.pkgmerge(" ".join([

"-s arch=sparc,{0}".format(self.rurl1),

"-s arch=i386,{0}".format(self.rurl2),

"-s arch=arm,{0}".format(self.rurl3),

"-d {0}".format(repodir),

"scheme amber@1.0 bronze@latest",

]))

# Get target repository catalog.

repo = self.get_repo(repodir)

cat = repo.get_catalog(pub="os.org")

# Verify the list of expected packages in the target repository.

expected = [

self.published[10], # pkg://os.org/amber@1.0-0

self.published[13], # pkg://os.org/bronze@2.0-0

self.published[5], # pkg://os.org/scheme@1.0-0

]

actual = [str(f) for f in sorted(cat.fmris())]

self.assertEqualDiff(expected, actual)

merged_expected["amber"] = """\

""".format(self.published[10]) # pkg://os.org/amber@1.0-0

# Verify that each package was merged correctly.

for f in cat.fmris():

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

expected = merged_expected[f.pkg_name]

self.assertEqualDiff(expected, actual)

# Cleanup.

shutil.rmtree(repodir)

def test_3_cross_merge(self):

"""Verify that pkgmerge works as expected when the resulting

# Create final merge repository.

repodir = os.path.join(self.test_root, "3merge_repo")

self.create_repo(repodir)

for i, arch in enumerate(("sparc", "i386", "arm")):

# For each arch, merge the debug and non-debug variants

# first.

rdir = os.path.join(self.test_root, "3{0}_repo".format(arch))

self.create_repo(rdir)

ndrepo = self.dcs[i + 1].get_repodir()

drepo = self.dcs[i + 4].get_repodir()

# Merge the packages.

self.pkgmerge(" ".join([

"-s debug=false,{0}".format(ndrepo),

"-s variant.debug=true,{0}".format(drepo),

"-d {0}".format(rdir)

]))

# Now merge all of the debug/non-debug repositories.

self.pkgmerge(" ".join([

"-s arch=sparc,{0}".format(os.path.join(self.test_root,

"3sparc_repo")),

"-s arch=i386,{0}".format(os.path.join(self.test_root,

"3i386_repo")),

"-s arch=arm,{0}".format(os.path.join(self.test_root,

"3arm_repo")),

"-d {0}".format(repodir),

]))

# Get target repository catalog.

repo = self.get_repo(repodir)

cat = repo.get_catalog(pub="os.org")

# Verify the list of expected packages in the target repository.

nlist = {}

for s in self.published + self.published_debug:

f = fmri.PkgFmri(s)

if f.pkg_name not in nlist or \

f.version > nlist[f.pkg_name].version:

nlist[f.pkg_name] = f

nlist = sorted(nlist.values())

expected = [str(f) for f in nlist]

actual = [str(f) for f in sorted(cat.fmris())]

self.assertEqualDiff(expected, actual)

# Verify that each package was merged correctly.

merged_expected = {

"amber": """\

""".format(self.published_debug[11]), # pkg://os.org/amber@2.0-0

"bronze": """\

""".format(self.published_debug[13]), # pkg://os.org/bronze@2.0-0

"scheme": """\

""".format(self.published_debug[5]), # pkg://os.org/scheme@1.0-0

"tree": """\

""".format(self.published_debug[14]), # pkg://os.org/tree@1.0-0

}

for f in nlist:

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

expected = merged_expected[f.pkg_name]

self.assertEqualDiff(expected, actual)

# Cleanup.

shutil.rmtree(repodir)

#

# Attempt to merge x86 and sparc debug/non-debug all at once.

#

self.create_repo(repodir)

self.pkgmerge(" ".join([

"-s arch=sparc,debug=false,{0}".format(self.dcs[1].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[2].get_repodir()),

"-s arch=arm,debug=false,{0}".format(self.dcs[3].get_repodir()),

"-s arch=sparc,debug=true,{0}".format(self.dcs[4].get_repodir()),

"-s arch=i386,debug=true,{0}".format(self.dcs[5].get_repodir()),

"-s arch=arm,debug=true,{0}".format(self.dcs[6].get_repodir()),

"-d {0}".format(repodir)

]))

repo = self.get_repo(repodir)

for f in nlist:

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

expected = merged_expected[f.pkg_name]

self.assertEqualDiff(expected, actual)

# Cleanup.

shutil.rmtree(repodir)

#

# Verify that naming an empty source allows a user to perform

# mismatched merges (i.e. merge sparc debug/non-debug with i386

# and arm non-debug) if empty repositories are used to fill in

# the missing cases.

nlist = {}

for s in self.published + self.published_debug[:6]:

f = fmri.PkgFmri(s)

if f.pkg_name not in nlist or \

f.version > nlist[f.pkg_name].version:

nlist[f.pkg_name] = f

nlist = sorted(nlist.values())

self.create_repo(repodir)

self.pkgmerge(" ".join([

"-s arch=sparc,debug=false,{0}".format(self.dcs[1].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[2].get_repodir()),

"-s arch=arm,debug=false,{0}".format(self.dcs[3].get_repodir()),

"-s arch=sparc,debug=true,{0}".format(self.dcs[4].get_repodir()),

# Explicitly state debug packages don't exist for these

# arch values by using an empty repository.

"-s arch=i386,debug=true,{0}".format(self.dcs[7].get_repodir()),

"-s arch=arm,debug=true,{0}".format(self.dcs[7].get_repodir()),

"-d {0}".format(repodir)

]))

# Verify that each package was merged correctly.

merged_expected = {

"amber": """\

""".format(self.published_debug[1]), # pkg://os.org/amber@2.0-0

"bronze": """\

""".format(self.published_debug[3]), # pkg://os.org/bronze@2.0-0

"scheme": """\

""".format(self.published_debug[5]), # pkg://os.org/scheme@1.0-0

"tree": """\

""".format(self.published_debug[4]), # pkg://os.org/tree@1.0-0

}

repo = self.get_repo(repodir)

for f in nlist:

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

expected = merged_expected[f.pkg_name]

self.assertEqualDiff(expected, actual)

# Cleanup.

shutil.rmtree(repodir)

def test_4_blend(self):

"""Make sure simple blending works"""

# Create the target repository.

repodir = os.path.join(self.test_root, "4merge_repo")

self.create_repo(repodir)

# Merge the silver packages.

self.pkgmerge(" ".join([

"-s arch=sparc,{0}".format(self.rurl8),

"-s arch=i386,{0}".format(self.rurl9),

"-d {0} silver".format(repodir)

]))

# get target repo

repo = self.get_repo(repodir)

cat = repo.get_catalog(pub="os.org")

expected = """\

""".format(self.published_blend[2])

for f in cat.fmris():

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

self.assertEqualDiff(expected, actual)

shutil.rmtree(repodir)

def test_5_blend(self):

"""test duplicate action detection during blending"""

repodir = os.path.join(self.test_root, "5merge_repo")

self.create_repo(repodir)

# Merge the tin packages - whoops

self.pkgmerge(" ".join([

"-s arch=sparc,{0}".format(self.rurl8),

"-s arch=i386,{0}".format(self.rurl9),

"-d {0} tin".format(repodir)

]), exit=1)

shutil.rmtree(repodir)

def test_6_blend(self):

"""check complex blending"""

repodir = os.path.join(self.test_root, "6merge_repo")

self.create_repo(repodir)

# merge the multi packages

self.pkgmerge(" ".join([

"-s arch=sparc,debug=true,{0}".format(self.dcs[10].get_repodir()),

"-s arch=i386,debug=true,{0}".format(self.dcs[12].get_repodir()),

"-s arch=sparc,debug=false,{0}".format(self.dcs[11].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[13].get_repodir()),

"-d {0}".format(repodir)]))

actual = self.get_manifest(repodir)

expected = """\

""".format(self.published_blend[-1])

self.assertEqualDiff(expected, actual)

shutil.rmtree(repodir)

def test_7_multipub_merge(self):

"""Tests that we can merge packages from repositories with

repodir = os.path.join(self.test_root, "7merge_repo")

self.create_repo(repodir)

# test dry run

self.pkgmerge(" ".join([

"-n",

"-s arch=sparc,debug=false,{0}".format(self.dcs[14].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[15].get_repodir()),

"-d {0}".format(repodir)]))

# test dry run with selected publishers

self.pkgmerge(" ".join([

"-p os.org",

"-p altpub",

"-n",

"-s arch=sparc,debug=false,{0}".format(self.dcs[14].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[15].get_repodir()),

"-d {0}".format(repodir)]))

# this should fail, as no -p noodles publisher exists in any of

# the source repositories

self.pkgmerge(" ".join([

"-p os.org",

"-p altpub",

"-p noodles",

"-n",

"-s arch=sparc,debug=false,{0}".format(self.dcs[14].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[15].get_repodir()),

"-d {0}".format(repodir)]), exit=1)

# now we want to perform the merge operations and validate the

# results. This was the order we published packages to multi_15

# 0 = pkg://os.org/scheme@1.0,5.11-0:20120920T085857Z

# 1 = pkg://os.org/tree@1.0,5.11-0:20120920T085859Z

# 2 = pkg://altpub/bronze@2.0,5.11-0:20120920T085902Z

# 3 = pkg://altpub/amber@1.0,5.11-0:20120920T085904Z

# 4 = pkg://last/gold@1.0,5.11-0:20120920T085906Z

# 5 = pkg://altpub/bronze@2.0,5.11-0:20120920T085920Z

# 6 = pkg://last/gold@1.0,5.11-0:20120920T085923Z

# build a dictionary of the FMRIs we're interested in

repo15_fmris = {

"osorg_scheme": self.published_multi_15[0],

"osorg_tree": self.published_multi_15[1],

"altpub_amber": self.published_multi_15[3],

# we published two versions of bronze and gold, use the

# latest FMRI

"altpub_bronze": self.published_multi_15[5],

"last_gold": self.published_multi_15[6]

}

# the some expected manifests we should get after merging.

expected_osorg_scheme = """\

""".format(**repo15_fmris)

expected_osorg_tree = """\

""".format(**repo15_fmris)

expected_altpub_amber = """\

""".format(**repo15_fmris)

expected_altpub_bronze = """\

""".format(**repo15_fmris)

expected_last_gold = """\

""".format(**repo15_fmris)

# A dictionary of the expected package contents, keyed by FMRI

expected = {

repo15_fmris["altpub_bronze"]: expected_altpub_bronze,

repo15_fmris["altpub_amber"]: expected_altpub_amber,

repo15_fmris["osorg_tree"]: expected_osorg_tree,

repo15_fmris["osorg_scheme"]: expected_osorg_scheme,

repo15_fmris["last_gold"]: expected_last_gold

}

def check_repo(repodir, keys, fmri_dic, expected):

"""Check that packages corresponding to the list of

sr = self.get_repo(repodir)

# check that the packages from 'keys' exist,

# and their content matches what we expect.

for key in keys:

f = fmri_dic[key]

with open(sr.manifest(f), "r") as manf:

actual = "".join(

sorted(l for l in manf)).strip()

self.assertEqualDiff(expected[f], actual)

# check that we have only the publishers used

# by packages from 'keys' in the repository

fmris = [fmri_dic[key] for key in keys]

pubs = set([fmri.PkgFmri(entry).get_publisher()

for entry in fmris])

known_pubs = set(

[p.prefix for p in sr.get_publishers()])

self.assertTrue(pubs == known_pubs,

"Repository at {0} didn't contain the "

"expected set of publishers")

# check that we have only the packages defined

# in 'keys' in the repository by walking all

# publishers, and all packages in the repository

for pub in sr.get_publishers():

cat = sr.get_catalog(pub=pub.prefix)

for f in cat.fmris():

if f.get_fmri() not in fmris:

self.assertTrue(False,

"{0} not in repository".format(f))

# test merging all publishers.

self.pkgmerge(" ".join([

"-s arch=sparc,debug=false,{0}".format(self.dcs[14].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[15].get_repodir()),

"-d {0}".format(repodir)]))

check_repo(repodir, repo15_fmris.keys(), repo15_fmris, expected)

# test merging only altpub and os.org.

shutil.rmtree(repodir)

self.create_repo(repodir)

self.pkgmerge(" ".join([

"-p altpub",

"-p os.org",

"-s arch=sparc,debug=false,{0}".format(self.dcs[14].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[15].get_repodir()),

"-d {0}".format(repodir)]))

check_repo(repodir, ["altpub_bronze", "altpub_amber",

"osorg_tree", "osorg_scheme"], repo15_fmris, expected)

# test merging only altpub

shutil.rmtree(repodir)

self.create_repo(repodir)

self.pkgmerge(" ".join([

"-p altpub",

"-s arch=sparc,debug=false,{0}".format(self.dcs[14].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[15].get_repodir()),

"-d {0}".format(repodir)]))

check_repo(repodir, ["altpub_bronze", "altpub_amber"],

repo15_fmris, expected)

# this should exit with a 1, but we should get the same results

# in the repository as last time.

shutil.rmtree(repodir)

self.create_repo(repodir)

self.pkgmerge(" ".join([

"-p altpub",

"-p noodles",

"-s arch=sparc,debug=false,{0}".format(self.dcs[14].get_repodir()),

"-s arch=i386,debug=false,{0}".format(self.dcs[15].get_repodir()),

"-d {0}".format(repodir)]), exit=1)

check_repo(repodir, ["altpub_bronze", "altpub_amber"],

repo15_fmris, expected)

def test_8_mediators(self):

"""test to make sure mediator-mediated links are not detected as collisions

# Create the target repository.

repodir = os.path.join(self.test_root, "8mediator_repo")

self.create_repo(repodir)

# Merge the two packages.

self.pkgmerge(" ".join([

"-s arch=PPC,{0}".format(self.rurl16),

"-s arch=ARM,{0}".format(self.rurl17),

"-d {0} mediator".format(repodir)

]))

# get target repo

repo = self.get_repo(repodir)

cat = repo.get_catalog(pub="os.org")

expected = """\

""".format(self.published_17[0])

for f in cat.fmris():

with open(repo.manifest(f), "r") as m:

actual = "".join(sorted(l for l in m)).strip()

self.assertEqualDiff(expected, actual)

shutil.rmtree(repodir)

def get_manifest(self, repodir, pubs=["os.org"]):

repository = self.get_repo(repodir)

actual = ""

for pub in pubs:

cat = repository.get_catalog(pub=pub)

for f in cat.fmris():

with open(repository.manifest(f), "r") as m:

actual += "".join(

sorted(l for l in m)).strip()

actual += "\n"

return actual.strip()

if __name__ == "__main__":

unittest.main()