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 tempfile

import time

import unittest

import zlib

from six.moves.urllib.parse import urlparse

from six.moves.urllib.request import url2pathname

class TestUtilMerge(pkg5unittest.ManyDepotTestCase):

# Cleanup after every test.

persistent_setup = False

scheme10 = """

tree10 = """

amber10 = """

amber20 = """

bronze10 = """

bronze20 = """

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

"tmp/bronze1", "tmp/bronze2",

"tmp/copyright2", "tmp/copyright3",

"tmp/libc.so.1", "tmp/sh"]

def setUp(self):

""" Start two depots.

# This test suite needs an actual depot.

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

start_depots=True)

self.make_misc_files(self.misc_files)

# Publish a set of packages to one repository.

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

self.durl1 = self.dcs[1].get_depot_url()

self.published = self.pkgsend_bulk(self.durl1, (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).

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

self.durl2 = self.dcs[2].get_depot_url()

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

self.amber20, self.bronze10, self.bronze20))

self.merge_dir = tempfile.mkdtemp(dir=self.test_root)

repo.repository_create(self.merge_dir)

@staticmethod

def get_repo(uri):

parts = urlparse(uri, "file", allow_fragments=0)

path = url2pathname(parts[2])

try:

return repo.Repository(root=path)

except cfg.ConfigError as e:

raise repo.RepositoryError(_("The specified "

"repository's configuration data is not "

"valid:\n{0}").format(e))

def test_0_merge(self):

"""Verify that merge functionality works as expected."""

pkg_names = set()

flist = []

for p in self.published:

f = fmri.PkgFmri(p)

pkg_names.add(f.pkg_name)

flist.append(f)

self.merge([

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

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

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

" ".join(pkg_names),

])

# Only get the newest FMRIs for each package.

flist.sort()

nlist = {}

for f in reversed(flist):

if f.pkg_name in nlist:

continue

nlist[f.pkg_name] = f

nlist = nlist.values()

def get_expected(f):

exp_lines = ["set name=pkg.fmri value={0}".format(f)]

for dc in self.dcs.values():

repo = dc.get_repo()

mpath = repo.manifest(f)

if not os.path.exists(mpath):

# Not in this repository, check next.

continue

m = open(mpath, "r")

for l in m:

if l.find("name=pkg.fmri") > -1:

continue

if l.find("name=variant") > -1:

continue

if not l.strip():

continue

exp_lines.append(l.strip())

m.close()

if f.pkg_name in ("tree", "scheme"):

# These packages are only published for sparc.

exp_lines.append("set name=variant.arch value=sparc")

else:

# Everything else is published for all variants.

exp_lines.append("set name=variant.arch value=sparc value=i386")

return "\n".join(sorted(exp_lines))

# Now load the manifest file for each package and verify that

# the merged manifest matches expectations.

for f in nlist:

mpath = os.path.join(self.merge_dir, "publisher",

"os.org", "pkg", f.get_dir_path())

m = open(mpath, "r")

returned = "".join(sorted(l for l in m))

returned = returned.strip()

m.close()

# Generate expected and verify.

expected = get_expected(f)

self.assertEqualDiff(expected, returned)

if __name__ == "__main__":

unittest.main()