1516N/A#!/usr/bin/python
581N/A#
581N/A# CDDL HEADER START
581N/A#
581N/A# The contents of this file are subject to the terms of the
581N/A# Common Development and Distribution License (the "License").
581N/A# You may not use this file except in compliance with the License.
581N/A#
581N/A# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
581N/A# or http://www.opensolaris.org/os/licensing.
581N/A# See the License for the specific language governing permissions
581N/A# and limitations under the License.
581N/A#
581N/A# When distributing Covered Code, include this CDDL HEADER in each
581N/A# file and include the License file at usr/src/OPENSOLARIS.LICENSE.
581N/A# If applicable, add the following below this CDDL HEADER, with the
581N/A# fields enclosed by brackets "[]" replaced with your own identifying
581N/A# information: Portions Copyright [yyyy] [name of copyright owner]
581N/A#
581N/A# CDDL HEADER END
581N/A#
581N/A
1899N/A#
3329N/A# Copyright (c) 2008, 2016, Oracle and/or its affiliates. All rights reserved.
1899N/A#
581N/A
3339N/Afrom . import testutils
581N/Aif __name__ == "__main__":
817N/A testutils.setup_environment("../../../proto")
1715N/Aimport pkg5unittest
581N/A
3130N/Aimport datetime
581N/Aimport os
581N/Aimport shutil
3234N/Aimport six
3339N/Aimport sys
817N/Aimport tempfile
1020N/Aimport time
1117N/Aimport unittest
3234N/A
3234N/Afrom six.moves import http_client
3234N/Afrom six.moves.urllib.error import HTTPError, URLError
3234N/Afrom six.moves.urllib.parse import quote, urljoin
3234N/Afrom six.moves.urllib.request import urlopen
581N/A
2028N/Aimport pkg.client.publisher as publisher
1507N/Aimport pkg.depotcontroller as dc
1507N/Aimport pkg.fmri as fmri
3130N/Aimport pkg.manifest as man
1507N/Aimport pkg.misc as misc
1968N/Aimport pkg.server.repository as sr
1507N/Aimport pkg.p5i as p5i
2992N/Aimport re
2992N/Aimport subprocess
581N/A
1715N/Aclass TestPkgDepot(pkg5unittest.SingleDepotTestCase):
581N/A # Only start/stop the depot once (instead of for every test)
1715N/A persistent_setup = True
581N/A
1151N/A foo10 = """
1151N/A open foo@1.0,5.11-0
3458N/A add dir path=foo/foo mode=0755 owner=root group=bin
1151N/A close """
1151N/A
1151N/A bar10 = """
1151N/A open bar@1.0,5.11-0
3458N/A add dir path=foo/bar mode=0755 owner=root group=bin
1151N/A close """
1151N/A
581N/A quux10 = """
581N/A open quux@1.0,5.11-0
581N/A add dir mode=0755 owner=root group=bin path=/bin
1715N/A add file tmp/cat mode=0555 owner=root group=bin path=/bin/cat
1715N/A add file tmp/libc.so.1 mode=0555 owner=root group=bin path=/lib/libc.so.1
581N/A close """
581N/A
689N/A info10 = """
689N/A open info@1.0,5.11-0
689N/A close """
689N/A
1025N/A update10 = """
1025N/A open update@1.0,5.11-0
1025N/A close """
1025N/A
1025N/A update11 = """
1025N/A open update@1.1,5.11-0
1025N/A close """
1025N/A
738N/A system10 = """
738N/A open system/libc@1.0,5.11-0
738N/A add set name="description" value="Package to test package names with slashes"
738N/A add dir path=tmp/foo mode=0755 owner=root group=bin
738N/A add depend type=require fmri=pkg:/SUNWcsl
738N/A close """
738N/A
2859N/A entire10 = """
2859N/A open entire@1.0,5.11-0
2859N/A add depend type=incorporate fmri=pkg:/foo
2859N/A close """
2859N/A
3130N/A info20 = """
3130N/A open info@2.0,5.11-0
3130N/A add set name="description" value="Test for checking info_0 consistency"
3130N/A add set name="pkg.human-version" value="test of human version"
3130N/A add dir mode=0755 owner=root group=bin path=/bin
3130N/A add file tmp/cat mode=0555 owner=root group=bin path=/bin/cat
3130N/A add file tmp/libc.so.1 mode=0555 owner=root group=bin path=/lib/libc.so.1
3130N/A close"""
3130N/A
1715N/A misc_files = [ "tmp/libc.so.1", "tmp/cat" ]
581N/A
581N/A def setUp(self):
3131N/A # test_info() parses dates,
3131N/A # so set expected locale before starting depots
3131N/A os.environ['LC_ALL'] = 'C'
3131N/A
1895N/A # This suite, for obvious reasons, actually needs a depot.
1895N/A pkg5unittest.SingleDepotTestCase.setUp(self, start_depot=True)
1715N/A self.make_misc_files(self.misc_files)
581N/A
581N/A def test_depot_ping(self):
581N/A """ Ping the depot several times """
581N/A
3339N/A self.assertTrue(self.dc.is_alive())
3339N/A self.assertTrue(self.dc.is_alive())
3339N/A self.assertTrue(self.dc.is_alive())
3339N/A self.assertTrue(self.dc.is_alive())
581N/A
581N/A def testStartStop(self):
581N/A """ Start and stop the depot several times """
581N/A self.dc.stop()
581N/A for i in range(0, 5):
581N/A self.dc.start()
3339N/A self.assertTrue(self.dc.is_alive())
581N/A self.dc.stop()
3339N/A self.assertTrue(not self.dc.is_alive())
581N/A
581N/A self.dc.start()
581N/A
581N/A def test_bug_1876(self):
581N/A """ Send package quux@1.0 an action at a time, restarting the
581N/A depot server after each one is sent, to ensure that
581N/A transactions work across depot restart. Then verify that
581N/A the package was successfully added by performing some
581N/A basic operations. """
581N/A
581N/A durl = self.dc.get_depot_url()
581N/A
581N/A for line in self.quux10.split("\n"):
581N/A line = line.strip()
581N/A if line == "":
581N/A continue
581N/A
581N/A try:
581N/A self.pkgsend(durl, line, exit = 0)
581N/A except:
581N/A self.pkgsend(durl, "close -A", exit = 0)
581N/A raise
581N/A
581N/A if not line == "close":
581N/A self.restart_depots()
581N/A
581N/A self.image_create(durl)
581N/A
581N/A self.pkg("list -a")
581N/A self.pkg("list", exit=1)
581N/A
581N/A self.pkg("install quux")
581N/A
581N/A self.pkg("list")
581N/A self.pkg("verify")
581N/A
581N/A self.pkg("uninstall quux")
581N/A self.pkg("verify")
581N/A
581N/A def test_bad_fmris(self):
581N/A durl = self.dc.get_depot_url()
581N/A self.pkgsend(durl, "open foo@", exit=1)
581N/A self.pkgsend(durl, "open foo@x.y", exit=1)
581N/A self.pkgsend(durl, "open foo@1.0,-2.0", exit=1)
581N/A
599N/A def test_bug_3365(self):
599N/A durl = self.dc.get_depot_url()
599N/A depotpath = self.dc.get_repodir()
599N/A
599N/A dir_file = os.path.join(depotpath, "search.dir")
599N/A pag_file = os.path.join(depotpath, "search.pag")
817N/A
3339N/A self.assertTrue(not os.path.exists(dir_file))
3339N/A self.assertTrue(not os.path.exists(pag_file))
599N/A
599N/A f = open(dir_file, "w")
599N/A f.close()
599N/A f = open(pag_file, "w")
599N/A f.close()
3339N/A self.assertTrue(os.path.exists(dir_file))
3339N/A self.assertTrue(os.path.exists(pag_file))
599N/A
599N/A self.dc.stop()
599N/A self.dc.start()
599N/A self.pkgsend_bulk(durl, self.quux10)
3339N/A self.assertTrue(not os.path.exists(dir_file))
3339N/A self.assertTrue(not os.path.exists(pag_file))
599N/A
689N/A def test_bug_4489(self):
689N/A """Publish a package and then verify that the depot /info
689N/A operation doesn't fail."""
689N/A depot_url = self.dc.get_depot_url()
689N/A plist = self.pkgsend_bulk(depot_url, self.info10)
3234N/A repourl = urljoin(depot_url, "info/0/{0}".format(plist[0]))
3234N/A urlopen(repourl)
581N/A
933N/A def test_bug_3739(self):
933N/A """Verify that a depot will return a 400 (Bad Request) error
933N/A whenever it is provided malformed FMRIs."""
933N/A
933N/A durl = self.dc.get_depot_url()
933N/A
933N/A for operation in ("info", "manifest"):
933N/A for entry in ("BRCMbnx", "BRCMbnx%40a",
933N/A "BRCMbnx%400.5.11%2C5.11-0.101%3A20081119T231649a"):
933N/A try:
3234N/A urlopen("{0}/{1}/0/{2}".format(durl,
933N/A operation, entry))
3234N/A except HTTPError as e:
3234N/A if e.code != http_client.BAD_REQUEST:
933N/A raise
933N/A
738N/A def test_bug_5366(self):
738N/A """Publish a package with slashes in the name, and then verify
738N/A that the depot manifest and info operations work regardless of
738N/A the encoding."""
738N/A depot_url = self.dc.get_depot_url()
738N/A plist = self.pkgsend_bulk(depot_url, self.system10)
738N/A # First, try it un-encoded.
3234N/A repourl = urljoin(depot_url, "info/0/{0}".format(plist[0]))
3234N/A urlopen(repourl)
3234N/A repourl = urljoin(depot_url, "manifest/0/{0}".format(
3158N/A plist[0]))
3234N/A urlopen(repourl)
738N/A # Second, try it encoded.
3234N/A repourl = urljoin(depot_url, "info/0/{0}".format(
3234N/A quote(plist[0])))
3234N/A urlopen(repourl)
3234N/A repourl = urljoin(depot_url, "manifest/0/{0}".format(
3234N/A quote(plist[0])))
3234N/A urlopen(repourl)
738N/A
3130N/A def test_info(self):
3130N/A """Testing information showed in /info/0."""
3130N/A
3130N/A depot_url = self.dc.get_depot_url();
3130N/A plist = self.pkgsend_bulk(depot_url, self.info20)
3130N/A
3234N/A openurl = urljoin(depot_url, "info/0/{0}".format(plist[0]))
3339N/A # urlopen.read return bytes
3339N/A content = misc.force_str(urlopen(openurl).read())
3130N/A # Get text from content.
3130N/A lines = content.splitlines()
3130N/A info_dic = {}
3130N/A attr_list = [
3130N/A 'Name',
3130N/A 'Summary',
3130N/A 'Publisher',
3130N/A 'Version',
3130N/A 'Build Release',
3130N/A 'Branch',
3130N/A 'Packaging Date',
3130N/A 'Size',
3130N/A 'Compressed Size',
3130N/A 'FMRI'
3130N/A ]
3130N/A
3130N/A for line in lines:
3339N/A fields = misc.force_str(line).split(":", 1)
3130N/A attr = fields[0].strip()
3130N/A if attr == "License":
3130N/A break
3130N/A if attr in attr_list:
3130N/A if len(fields) == 2:
3130N/A info_dic[attr] = fields[1].strip()
3130N/A else:
3130N/A info_dic[attr] = ""
3130N/A
3130N/A # Read manifest.
3234N/A openurl = urljoin(depot_url, "manifest/0/{0}".format(plist[0]))
3339N/A content = misc.force_str(urlopen(openurl).read())
3130N/A manifest = man.Manifest()
3130N/A manifest.set_content(content=content)
3130N/A fmri_content = manifest.get("pkg.fmri", "")
3130N/A
3130N/A # Check if FMRI is empty.
3339N/A self.assertTrue(fmri_content)
3130N/A pfmri = fmri.PkgFmri(fmri_content, None)
3130N/A pub, name, ver = pfmri.tuple()
3130N/A size, csize = manifest.get_size()
3130N/A
3130N/A # Human version.
3130N/A version = info_dic['Version']
3130N/A hum_ver = ""
3130N/A if '(' in version:
3130N/A start = version.find('(')
3130N/A end = version.rfind(')')
3130N/A hum_ver = version[start + 1 : end - len(version)]
3130N/A version = version[:start - len(version)]
3130N/A info_dic['Version'] = version.strip()
3130N/A
3130N/A # Compare each attribute.
3130N/A self.assertEqual(info_dic["Summary"], manifest.get("pkg.summary",
3130N/A manifest.get("description", "")))
3130N/A self.assertEqual(info_dic["Version"], str(ver.release))
3130N/A self.assertEqual(hum_ver, manifest.get("pkg.human-version",""))
3130N/A self.assertEqual(info_dic["Name"], name)
3130N/A self.assertEqual(info_dic["Publisher"], pub)
3130N/A self.assertEqual(info_dic["Build Release"], str(ver.build_release))
3130N/A timestamp = datetime.datetime.strptime(
3131N/A info_dic["Packaging Date"], "%a %b %d %H:%M:%S %Y")
3130N/A self.assertEqual(timestamp, ver.get_timestamp())
3130N/A self.assertEqual(info_dic["Size"], misc.bytes_to_str(size))
3130N/A self.assertEqual(info_dic["Compressed Size"], misc.bytes_to_str(csize))
3130N/A self.assertEqual(info_dic["FMRI"], fmri_content)
3339N/A if six.PY3:
3339N/A os.environ["LC_ALL"] = "en_US.UTF-8"
3130N/A
1151N/A def test_bug_5707(self):
1151N/A """Testing depotcontroller.refresh()."""
1151N/A
1151N/A depot_url = self.dc.get_depot_url()
1151N/A self.pkgsend_bulk(depot_url, self.foo10)
1151N/A
1151N/A self.image_create(depot_url)
1151N/A self.pkg("install foo")
1151N/A self.pkg("verify")
1151N/A
3158N/A depot_file_url = "file://{0}".format(self.dc.get_repodir())
1151N/A self.pkgsend_bulk(depot_url, self.bar10)
1151N/A self.pkg("refresh")
1151N/A
1151N/A self.pkg("install bar")
1151N/A self.pkg("verify")
1151N/A
1151N/A self.dc.refresh()
1151N/A self.pkg("refresh")
1151N/A
1505N/A self.pkg("install bar", exit=4) # nothing to do
1151N/A self.pkg("verify")
1151N/A
742N/A def test_face_root(self):
742N/A """Verify that files outside of the package content web root
742N/A cannot be accessed, and that files inside can be."""
742N/A depot_url = self.dc.get_depot_url()
742N/A # Since /usr/share/lib/pkg/web/ is the content web root,
742N/A # any attempts to go outside that directory should fail
742N/A # with a 404 error.
742N/A try:
3234N/A urlopen("{0}/../../../../bin/pkg".format(depot_url))
3234N/A except HTTPError as e:
3234N/A if e.code != http_client.NOT_FOUND:
742N/A raise
742N/A
3234N/A f = urlopen("{0}/robots.txt".format(depot_url))
3339N/A self.assertTrue(len(f.read()))
742N/A f.close()
742N/A
873N/A def test_repo_create(self):
873N/A """Verify that starting a depot server in readonly mode with
1672N/A a non-existent or empty repo_dir fails and that permissions
1672N/A errors are handled correctly during creation. Then verify
1672N/A that starting a depot with the same directory in publishing
1672N/A mode works and then a readonly depot again after that works.
1672N/A """
873N/A
1715N/A dpath = os.path.join(self.test_root, "repo_create")
873N/A
873N/A opath = self.dc.get_repodir()
873N/A self.dc.set_repodir(dpath)
873N/A
873N/A # First, test readonly mode with a repo_dir that doesn't exist.
873N/A self.dc.set_readonly()
873N/A self.dc.stop()
873N/A self.dc.start_expected_fail()
3339N/A self.assertTrue(not self.dc.is_alive())
873N/A
873N/A # Next, test readonly mode with a repo_dir that is empty.
1507N/A os.makedirs(dpath, misc.PKG_DIR_MODE)
873N/A self.dc.set_readonly()
873N/A self.dc.start_expected_fail()
3339N/A self.assertTrue(not self.dc.is_alive())
873N/A
873N/A # Next, test readwrite (publishing) mode with a non-existent
873N/A # repo_dir.
873N/A shutil.rmtree(dpath)
873N/A self.dc.set_readwrite()
873N/A self.dc.start()
3339N/A self.assertTrue(self.dc.is_alive())
873N/A self.dc.stop()
3339N/A self.assertTrue(not self.dc.is_alive())
873N/A
1672N/A # Next, test readwrite (publishing) mode with a non-existent
1672N/A # repo_dir for an unprivileged user.
1672N/A shutil.rmtree(dpath)
1672N/A self.dc.set_readwrite()
1672N/A wr_start, wr_end = self.dc.get_wrapper()
1672N/A su_wrap, su_end = self.get_su_wrapper(su_wrap=True)
1672N/A try:
1672N/A self.dc.set_wrapper([su_wrap], su_end)
1672N/A self.dc.start_expected_fail(exit=1)
1672N/A finally:
1672N/A # Even if this test fails, this wrapper must be reset.
1672N/A self.dc.set_wrapper(wr_start, wr_end)
3339N/A self.assertTrue(not self.dc.is_alive())
1672N/A
873N/A # Next, test readwrite (publishing) mode with an empty repo_dir.
1507N/A os.makedirs(dpath, misc.PKG_DIR_MODE)
873N/A self.dc.set_readwrite()
873N/A self.dc.start()
3339N/A self.assertTrue(self.dc.is_alive())
873N/A self.dc.stop()
3339N/A self.assertTrue(not self.dc.is_alive())
873N/A
873N/A # Finally, re-test readonly mode now that the repository has
873N/A # been created.
873N/A self.dc.set_readonly()
873N/A self.dc.start()
3339N/A self.assertTrue(self.dc.is_alive())
873N/A self.dc.stop()
3339N/A self.assertTrue(not self.dc.is_alive())
873N/A
873N/A # Cleanup.
873N/A shutil.rmtree(dpath)
873N/A self.dc.set_repodir(opath)
873N/A
2026N/A def test_append_reopen(self):
2026N/A """Test that if a depot has a partially finished append
2026N/A transaction, that it reopens it correctly."""
2026N/A
2026N/A durl = self.dc.get_depot_url()
2026N/A plist = self.pkgsend_bulk(durl, self.foo10)
3158N/A self.pkgsend(durl, "append {0}".format(plist[0]))
2026N/A self.dc.stop()
2026N/A self.dc.start()
2026N/A self.pkgsend(durl, "close")
2026N/A
2026N/A def test_nonsig_append(self):
2026N/A """Test that sending a non-signature action to an append
2026N/A transaction results in an error."""
2026N/A
2026N/A durl = self.dc.get_depot_url()
2026N/A plist = self.pkgsend_bulk(durl, self.foo10)
3158N/A self.pkgsend(durl, "append {0}".format(plist[0]))
2026N/A self.pkgsend(durl, "add dir path=tmp/foo mode=0755 "
2026N/A "owner=root group=bin", exit=1)
2026N/A
2044N/A def test_root_link(self):
2044N/A """Verify that the depot server accepts a link to a
2044N/A directory as a repository root."""
2044N/A
2044N/A if self.dc.started:
2044N/A self.dc.stop()
2044N/A
2044N/A # Create a link to the repository and verify that
2044N/A # the depot server allows it.
2044N/A lsrc = self.dc.get_repodir()
2044N/A ltarget = os.path.join(self.test_root, "depot_link")
2044N/A os.symlink(lsrc, ltarget)
2044N/A self.dc.set_repodir(ltarget)
2044N/A self.dc.start()
2044N/A
2044N/A # Reset for any tests that might execute afterwards.
2044N/A os.unlink(ltarget)
2044N/A self.dc.stop()
2044N/A self.dc.set_repodir(lsrc)
2044N/A self.dc.start()
2044N/A
2859N/A def test_empty_incorp_depend(self):
2859N/A """ Bug 16304629
2859N/A Test that a version-less incorporate dependency in a package
2859N/A doesn't cause a traceback and a 404 in the BUI.
2859N/A """
2859N/A depot_url = self.dc.get_depot_url()
2859N/A self.pkgsend_bulk(depot_url, self.foo10)
2859N/A self.pkgsend_bulk(depot_url, self.entire10)
2859N/A
3234N/A repourl = urljoin(depot_url,
3158N/A "/en/catalog.shtml?version={0}&action=Browse".format(
3234N/A quote("entire@1.0,5.11-0")))
2859N/A
3234N/A res = urlopen(repourl)
2859N/A
3274N/A def test_publisher_prefix(self):
3274N/A """Test that various publisher prefixes can be understood
3274N/A by CherryPy's dispatcher."""
3274N/A
3274N/A if self.dc.started:
3274N/A self.dc.stop()
3274N/A
3274N/A depot_url = self.dc.get_depot_url()
3274N/A repopath = os.path.join(self.test_root, "repo")
3274N/A self.create_repo(repopath)
3274N/A self.dc.set_repodir(repopath)
3274N/A pubs = ["test-hyphen", "test.dot"]
3274N/A for p in pubs:
3274N/A self.pkgrepo("-s {0} add-publisher {1}".format(
3274N/A repopath, p))
3274N/A self.dc.start()
3274N/A for p in pubs:
3274N/A # test that the catalog file can be found
3274N/A url = urljoin(depot_url,
3274N/A "{0}/catalog/1/catalog.attrs".format(p))
3274N/A urlopen(url)
3274N/A
873N/A
1715N/Aclass TestDepotController(pkg5unittest.CliTestCase):
581N/A
581N/A def setUp(self):
1715N/A pkg5unittest.CliTestCase.setUp(self)
581N/A
581N/A self.__dc = dc.DepotController()
581N/A self.__pid = os.getpid()
1431N/A self.__dc.set_property("publisher", "prefix", "test")
3049N/A self.__dc.set_depotd_path(pkg5unittest.g_pkg_path + \
581N/A "/usr/lib/pkg.depotd")
3049N/A self.__dc.set_depotd_content_root(pkg5unittest.g_pkg_path + \
612N/A "/usr/share/lib/pkg")
581N/A
2028N/A repopath = os.path.join(self.test_root, "repo")
1715N/A logpath = os.path.join(self.test_root, self.id())
2028N/A self.create_repo(repopath, properties={ "publisher": {
2028N/A "prefix": "test" }})
2028N/A self.__dc.set_repodir(repopath)
2028N/A self.__dc.set_logpath(logpath)
581N/A
2028N/A def _get_repo_index_dir(self):
2028N/A depotpath = self.__dc.get_repodir()
2028N/A repo = self.__dc.get_repo()
2028N/A rstore = repo.get_pub_rstore("test")
2028N/A return rstore.index_root
581N/A
2028N/A def _get_repo_writ_dir(self):
2028N/A depotpath = self.__dc.get_repodir()
2028N/A repo = self.__dc.get_repo()
2028N/A rstore = repo.get_pub_rstore("test")
2028N/A return rstore.writable_root
581N/A
581N/A def tearDown(self):
1715N/A pkg5unittest.CliTestCase.tearDown(self)
581N/A self.__dc.kill()
581N/A
581N/A def testStartStop(self):
2272N/A self.__dc.set_port(self.next_free_port)
581N/A for i in range(0, 5):
581N/A self.__dc.start()
3339N/A self.assertTrue(self.__dc.is_alive())
581N/A self.__dc.stop()
3339N/A self.assertTrue(not self.__dc.is_alive())
581N/A
797N/A def test_cfg_file(self):
1715N/A cfg_file = os.path.join(self.test_root, "cfg2")
797N/A fh = open(cfg_file, "w")
797N/A fh.close()
2272N/A self.__dc.set_port(self.next_free_port)
797N/A self.__dc.set_cfg_file(cfg_file)
797N/A self.__dc.start()
797N/A
975N/A def test_writable_root(self):
975N/A """Tests whether the index and feed cache file are written to
975N/A the writable root parameter."""
1899N/A
1715N/A self.make_misc_files(TestPkgDepot.misc_files)
1715N/A writable_root = os.path.join(self.test_root,
975N/A "writ_root")
2028N/A o_index_dir = os.path.join(self._get_repo_index_dir(), "index")
975N/A
1020N/A timeout = 10
1507N/A
975N/A def check_state(check_feed):
2028N/A index_dir = os.path.join(self._get_repo_writ_dir(),
2028N/A "index")
2028N/A feed = os.path.join(writable_root, "publisher", "test",
2028N/A "feed.xml")
1020N/A found = not os.path.exists(o_index_dir) and \
1020N/A os.path.isdir(index_dir) and \
1020N/A (not check_feed or os.path.isfile(feed))
1020N/A start_time = time.time()
1020N/A while not found and time.time() - start_time < timeout:
1020N/A time.sleep(1)
1020N/A found = not os.path.exists(o_index_dir) and \
1020N/A os.path.isdir(index_dir) and \
1020N/A (not check_feed or os.path.isfile(feed))
1020N/A
3339N/A self.assertTrue(not os.path.exists(o_index_dir))
3339N/A self.assertTrue(os.path.isdir(index_dir))
975N/A if check_feed:
2028N/A try:
3339N/A self.assertTrue(os.path.isfile(feed))
2028N/A except:
2028N/A raise RuntimeError("Feed cache file "
3158N/A "not found at '{0}'.".format(feed))
2028N/A def get_feed(durl, pub=""):
1020N/A start_time = time.time()
1020N/A got = False
1020N/A while not got and (time.time() - start_time) < timeout:
2028N/A if pub:
3158N/A pub = "{0}/".format(pub)
1020N/A try:
3234N/A urlopen("{0}{1}/feed".format(durl,
2028N/A pub))
1020N/A got = True
3234N/A except HTTPError as e:
1968N/A self.debug(str(e))
1020N/A time.sleep(1)
3339N/A self.assertTrue(got)
1507N/A
2272N/A self.__dc.set_port(self.next_free_port)
2028N/A durl = self.__dc.get_depot_url()
2028N/A
2028N/A repo = self.__dc.get_repo()
2028N/A pub = repo.get_publisher("test")
2310N/A pub_repo = pub.repository
2028N/A if not pub_repo:
2028N/A pub_repo = publisher.Repository()
2310N/A pub.repository = pub_repo
2028N/A pub_repo.origins = [durl]
2028N/A repo.update_publisher(pub)
2028N/A
975N/A self.__dc.set_writable_root(writable_root)
1736N/A self.__dc.set_property("publisher", "prefix", "test")
975N/A self.__dc.start()
975N/A check_state(False)
2028N/A self.pkgsend_bulk(durl, TestPkgDepot.quux10, refresh_index=True)
1020N/A get_feed(durl)
975N/A check_state(True)
975N/A
975N/A self.image_create(durl)
975N/A self.pkg("search -r cat")
975N/A self.__dc.stop()
975N/A self.__dc.set_readonly()
975N/A shutil.rmtree(writable_root)
975N/A self.__dc.start()
1020N/A get_feed(durl)
1020N/A check_state(True)
1020N/A self.pkg("search -r cat")
1020N/A self.__dc.stop()
1020N/A self.__dc.set_refresh_index()
1020N/A shutil.rmtree(writable_root)
1020N/A self.__dc.start()
1020N/A check_state(False)
1020N/A self.__dc.stop()
1020N/A self.__dc.set_norefresh_index()
1020N/A self.__dc.start()
1020N/A get_feed(durl)
975N/A check_state(True)
975N/A self.pkg("search -r cat")
975N/A
581N/A def testBadArgs(self):
2272N/A self.__dc.set_port(self.next_free_port)
581N/A self.__dc.set_readonly()
581N/A self.__dc.set_rebuild()
581N/A self.__dc.set_norefresh_index()
581N/A
3339N/A self.assertTrue(self.__dc.start_expected_fail())
581N/A
581N/A self.__dc.set_readonly()
581N/A self.__dc.set_norebuild()
581N/A self.__dc.set_refresh_index()
581N/A
3339N/A self.assertTrue(self.__dc.start_expected_fail())
581N/A
581N/A self.__dc.set_readonly()
581N/A self.__dc.set_rebuild()
581N/A self.__dc.set_refresh_index()
581N/A
3339N/A self.assertTrue(self.__dc.start_expected_fail())
581N/A
581N/A self.__dc.set_readwrite()
581N/A self.__dc.set_rebuild()
581N/A self.__dc.set_refresh_index()
581N/A
3339N/A self.assertTrue(self.__dc.start_expected_fail())
581N/A
581N/A self.__dc.set_mirror()
581N/A self.__dc.set_rebuild()
581N/A self.__dc.set_norefresh_index()
581N/A
3339N/A self.assertTrue(self.__dc.start_expected_fail())
581N/A
581N/A self.__dc.set_mirror()
581N/A self.__dc.set_norebuild()
581N/A self.__dc.set_refresh_index()
581N/A
3339N/A self.assertTrue(self.__dc.start_expected_fail())
581N/A
1431N/A def test_disable_ops(self):
1431N/A """Verify that disable-ops works as expected."""
1431N/A
1431N/A # For this disabled case, /catalog/1/ should return
1431N/A # a NOT_FOUND error.
1431N/A self.__dc.set_disable_ops(["catalog/1"])
2272N/A self.__dc.set_port(self.next_free_port)
1431N/A self.__dc.start()
1431N/A durl = self.__dc.get_depot_url()
1431N/A try:
3234N/A urlopen("{0}/catalog/1/".format(durl))
3234N/A except HTTPError as e:
3234N/A self.assertEqual(e.code, http_client.NOT_FOUND)
1431N/A self.__dc.stop()
1431N/A
1431N/A # For this disabled case, all /catalog/ operations should return
1431N/A # a NOT_FOUND error.
1431N/A self.__dc.set_disable_ops(["catalog"])
2272N/A self.__dc.set_port(self.next_free_port)
1431N/A self.__dc.start()
1431N/A durl = self.__dc.get_depot_url()
1431N/A for ver in (0, 1):
1431N/A try:
3234N/A urlopen("{0}/catalog/{1:d}/".format(durl, ver))
3234N/A except HTTPError as e:
3234N/A self.assertEqual(e.code, http_client.NOT_FOUND)
1431N/A self.__dc.stop()
1431N/A
1431N/A # In the normal case, /catalog/1/ should return
1431N/A # a FORBIDDEN error.
1431N/A self.__dc.unset_disable_ops()
1431N/A self.__dc.start()
1431N/A durl = self.__dc.get_depot_url()
1431N/A try:
3234N/A urlopen("{0}/catalog/1/".format(durl))
3234N/A except HTTPError as e:
3234N/A self.assertEqual(e.code, http_client.FORBIDDEN)
1431N/A self.__dc.stop()
1431N/A
1431N/A # A bogus operation should prevent the depot from starting.
1431N/A self.__dc.set_disable_ops(["no_such_op/0"])
1431N/A self.__dc.start_expected_fail()
1431N/A self.assertFalse(self.__dc.is_alive())
1431N/A
817N/A
1715N/Aclass TestDepotOutput(pkg5unittest.SingleDepotTestCase):
1117N/A # Since these tests are output sensitive, the depots should be purged
1117N/A # after each one is run.
1715N/A persistent_setup = False
817N/A
817N/A quux10 = """
817N/A open quux@1.0,5.11-0
817N/A add dir mode=0755 owner=root group=bin path=/bin
817N/A close """
817N/A
817N/A info10 = """
817N/A open info@1.0,5.11-0
817N/A close """
817N/A
1899N/A file10 = """
1899N/A open file@1.0,5.11-0
1899N/A add dir mode=0755 owner=root group=bin path=/var
1899N/A add file tmp/file path=var/file mode=644 owner=root group=bin
1899N/A close """
1899N/A
817N/A system10 = """
817N/A open system/libc@1.0,5.11-0
817N/A add set name="description" value="Package to test package names with slashes"
817N/A add dir path=tmp/foo mode=0755 owner=root group=bin
817N/A add depend type=require fmri=pkg:/SUNWcsl
817N/A close """
817N/A
1117N/A zfsextras10 = """
1117N/A open zfs-extras@1.0,5.11-0
1117N/A close """
817N/A
1117N/A zfsutils10 = """
1117N/A open zfs/utils@1.0,5.11-0
1117N/A close """
817N/A
1117N/A repo_cfg = {
1117N/A "publisher": {
1117N/A "prefix": "org.opensolaris.pending"
1117N/A },
2028N/A }
2028N/A
2028N/A pub_repo_cfg = {
2028N/A "collection_type": "supplemental",
2028N/A "description":
2028N/A "Development packages for the contrib repository.",
2028N/A "legal_uris": [
2028N/A "http://www.opensolaris.org/os/copyrights",
2028N/A "http://www.opensolaris.org/os/tou",
2028N/A "http://www.opensolaris.org/os/trademark"
2028N/A ],
2028N/A "mirrors": [],
2028N/A "name": """"Pending" Repository""",
2028N/A "origins": [], # Has to be set during setUp for correct origin.
2028N/A "refresh_seconds": 86400,
2028N/A "registration_uri": "",
2028N/A "related_uris": [
2028N/A "http://jucr.opensolaris.org/contrib",
2028N/A "http://jucr.opensolaris.org/pending",
2028N/A "http://pkg.opensolaris.org/contrib",
2028N/A ]
1117N/A }
817N/A
1117N/A def setUp(self):
2028N/A pkg5unittest.SingleDepotTestCase.setUp(self)
1117N/A
1431N/A # Prevent override of custom configuration;
1431N/A # tests will set as needed.
1431N/A self.dc.clear_property("publisher", "prefix")
1431N/A
1715N/A self.tpath = tempfile.mkdtemp(prefix="tpath",
1715N/A dir=self.test_root)
817N/A
1899N/A self.make_misc_files("tmp/file")
1899N/A
2992N/A def __depot_daemon_start(self, repopath, out_path, err_path):
2992N/A """Helper function: start a depot daemon and return a handler
2992N/A for its parent process and the port number it is running on.
2992N/A The parent process is a ctrun process. It is the user's
2992N/A responsibility to call depot_daemon_stop to stop the process.
2992N/A ctrun is used to kill the depot daemon process after finishing
2992N/A testing. It is needed because the double fork machanism of the
2992N/A daemonizer make the daemonized depot server indenpendent from
2992N/A its parent process.
2992N/A
2992N/A repopath: The repository path that a depot daemon will run on.
2992N/A
2992N/A out_path: The depot daemon stdout log file path.
2992N/A
2992N/A err_path: The depot daemon stderr log file path."""
2992N/A
2992N/A # Make sure the PKGDEPOT_CONTROLLER is not set in the newenv.
2992N/A newenv = os.environ.copy()
2992N/A newenv.pop("PKGDEPOT_CONTROLLER", None)
3339N/A cmdargs = "/usr/bin/ctrun -o noorphan {0} {1}/usr/lib/pkg.depotd " \
3339N/A "-p {2} -d {3} --content-root {4}/usr/share/lib/pkg " \
3339N/A "--readonly </dev/null --log-access={5} " \
3339N/A "--log-errors={6}".format(sys.executable,
3339N/A pkg5unittest.g_pkg_path, self.next_free_port, repopath,
3339N/A pkg5unittest.g_pkg_path, out_path, err_path)
2992N/A
2992N/A curport = self.next_free_port
2992N/A self.next_free_port += 1
2992N/A
2992N/A # Start a depot daemon process.
2992N/A try:
2992N/A depot_handle = subprocess.Popen(cmdargs, env=newenv,
2992N/A shell=True)
2992N/A
3339N/A self.assertTrue(depot_handle != None, msg="Could not "
2992N/A "start depot")
2992N/A begintime = time.time()
2992N/A check_interval = 0.20
2992N/A daemon_started = False
3158N/A durl = "http://localhost:{0}/en/index.shtml".format(curport)
2992N/A
2992N/A while (time.time() - begintime) <= 40.0:
2992N/A rc = depot_handle.poll()
3339N/A self.assertTrue(rc is None, msg="Depot exited "
2992N/A "unexpectedly")
2992N/A
2992N/A try:
3234N/A f = urlopen(durl)
2992N/A daemon_started = True
2992N/A break
3234N/A except URLError as e:
2992N/A time.sleep(check_interval)
2992N/A
2992N/A if not daemon_started:
2992N/A if depot_handle:
2992N/A depot_handle.kill()
3339N/A self.assertTrue(daemon_started, msg="Could not "
2992N/A "access depot daemon")
2992N/A
2992N/A # Read the msgs from the err log file to verify log
2992N/A # msgs.
2992N/A with open(err_path, "r") as read_err:
2992N/A msgs = read_err.readlines()
3339N/A self.assertTrue(msgs, "Log message is "
2992N/A "empty. Check if the previous "
2992N/A "ctrun process shut down "
2992N/A "properly")
2992N/A return (depot_handle, curport)
2992N/A except:
2992N/A try:
2992N/A if depot_handle:
2992N/A depot_handle.kill()
2992N/A except:
2992N/A pass
2992N/A raise
2992N/A
2992N/A def __depot_daemon_stop(self, depot_handle):
2992N/A """Helper function: stop the depot daemon process by sending
2992N/A signal to its handle."""
2992N/A
2992N/A # Terminate the depot daemon. If failed, kill it.
2992N/A try:
2992N/A if depot_handle:
2992N/A depot_handle.terminate()
2992N/A except:
2992N/A try:
2992N/A if depot_handle:
2992N/A depot_handle.kill()
2992N/A except:
2992N/A pass
2992N/A
817N/A def test_0_depot_bui_output(self):
817N/A """Verify that a non-error response and valid HTML is returned
817N/A for each known BUI page in every available depot mode."""
817N/A
2028N/A pub = "test"
2028N/A self.dc.set_property("publisher", "prefix", pub)
1431N/A
817N/A # A list of tuples containing the name of the method used to set
817N/A # the mode, and then the method needed to unset that mode.
817N/A mode_methods = [
817N/A ("set_readwrite", None),
817N/A ("set_mirror", "unset_mirror"),
817N/A ("set_readonly", "set_readwrite"),
817N/A ]
817N/A
817N/A pages = [
817N/A "index.shtml",
817N/A "en/catalog.shtml",
817N/A "en/index.shtml",
1116N/A "en/advanced_search.shtml",
817N/A "en/search.shtml",
817N/A "en/stats.shtml",
817N/A ]
817N/A
2028N/A repodir = self.dc.get_repodir()
2028N/A durl = self.dc.get_depot_url()
817N/A for with_packages in (False, True):
2028N/A shutil.rmtree(repodir, ignore_errors=True)
2028N/A
2028N/A # Create repository and set publisher origins.
2028N/A self.create_repo(self.dc.get_repodir())
3158N/A self.pkgrepo("set -s {repodir} -p {pub} "
3158N/A "repository/origins={durl}".format(**locals()))
817N/A
817N/A if with_packages:
1117N/A self.dc.set_readwrite()
1117N/A self.dc.start()
1895N/A self.pkgsend_bulk(durl, (self.info10,
1895N/A self.quux10, self.system10))
1117N/A self.dc.stop()
817N/A
817N/A for set_method, unset_method in mode_methods:
817N/A if set_method:
1117N/A getattr(self.dc, set_method)()
817N/A
1117N/A self.dc.start()
817N/A for path in pages:
817N/A # Any error responses will cause an
817N/A # exception.
3234N/A response = urlopen(
3158N/A "{0}/{1}".format(durl, path))
817N/A
817N/A fd, fpath = tempfile.mkstemp(
817N/A suffix="html", dir=self.tpath)
3339N/A fp = os.fdopen(fd, "wb")
817N/A fp.write(response.read())
817N/A fp.close()
817N/A
2603N/A # Because the 'role' attribute used for
3130N/A # screen readers and other accessibility
2603N/A # tools isn't part of the official XHTML
2603N/A # 1.x standards, it has to be dropped
2603N/A # for the document to be validated.
2603N/A # Setting 'drop_prop_attrs' to True here
3130N/A # does that while ensuring that the
2603N/A # output of the depot is otherwise
2603N/A # standards-compliant.
2603N/A self.validate_html_file(fpath,
2603N/A drop_prop_attrs=True)
817N/A
1117N/A self.dc.stop()
817N/A if unset_method:
1117N/A getattr(self.dc, unset_method)()
1117N/A
1117N/A def __update_repo_config(self):
1117N/A """Helper function to generate test repository configuration."""
1117N/A # Find and load the repository configuration.
1117N/A rpath = self.dc.get_repodir()
2028N/A assert os.path.isdir(rpath)
1117N/A rcpath = os.path.join(rpath, "cfg_cache")
1117N/A
1968N/A rc = sr.RepositoryConfig(target=rcpath)
1117N/A
1117N/A # Update the configuration with our sample data.
1968N/A cfgdata = self.repo_cfg
1968N/A for section in cfgdata:
1968N/A for prop in cfgdata[section]:
1431N/A rc.set_property(section, prop,
1968N/A cfgdata[section][prop])
1117N/A
1117N/A # Save it.
1847N/A rc.write()
1117N/A
2028N/A # Apply publisher properties and update.
2028N/A repo = self.dc.get_repo()
2028N/A try:
2028N/A pub = repo.get_publisher("org.opensolaris.pending")
2028N/A except sr.RepositoryUnknownPublisher:
2028N/A pub = publisher.Publisher("org.opensolaris.pending")
2028N/A repo.add_publisher(pub)
2028N/A
2310N/A pub_repo = pub.repository
2028N/A if not pub_repo:
2028N/A pub_repo = publisher.Repository()
2310N/A pub.repository = pub_repo
2028N/A
3234N/A for attr, val in six.iteritems(self.pub_repo_cfg):
2028N/A setattr(pub_repo, attr, val)
2028N/A repo.update_publisher(pub)
2028N/A
1117N/A def test_1_depot_publisher(self):
1117N/A """Verify the output of the depot /publisher operation."""
1117N/A
1117N/A # Now update the repository configuration while the depot is
1117N/A # stopped so changes won't be overwritten on exit.
1117N/A self.__update_repo_config()
1117N/A
1117N/A # Start the depot.
1117N/A self.dc.start()
1117N/A
1117N/A durl = self.dc.get_depot_url()
3234N/A purl = urljoin(durl, "publisher/0")
2028N/A entries = p5i.parse(location=purl)
3339N/A # entries's order is unstable in Python 3, but it doesn't really
3339N/A # matter as long as the prefix is in entries.
3339N/A if entries[0][0].prefix == "test":
3339N/A index = -1
3339N/A assert entries[1][0].prefix == "org.opensolaris.pending"
3339N/A else:
3339N/A index = 0
3339N/A assert entries[0][0].prefix == "org.opensolaris.pending"
3339N/A assert entries[1][0].prefix == "test"
3339N/A
1117N/A
1117N/A # Now verify that the parsed response has the expected data.
3339N/A pub, pkglist = entries[index]
1968N/A cfgdata = self.repo_cfg
1968N/A for prop in cfgdata["publisher"]:
1431N/A self.assertEqual(getattr(pub, prop),
1968N/A cfgdata["publisher"][prop])
1117N/A
2310N/A repo = pub.repository
3234N/A for prop, expected in six.iteritems(self.pub_repo_cfg):
1431N/A returned = getattr(repo, prop)
1431N/A if prop.endswith("uris") or prop == "origins":
1117N/A uris = []
1117N/A for u in returned:
1736N/A uri = u.uri
1736N/A if uri.endswith("/"):
1736N/A uri = uri[:-1]
1736N/A uris.append(uri)
1117N/A returned = uris
1117N/A self.assertEqual(returned, expected)
1117N/A
1117N/A def test_2_depot_p5i(self):
3274N/A """Verify the output of the depot /p5i operation."""
1117N/A
1117N/A # Now update the repository configuration while the depot is
1117N/A # stopped so changes won't be overwritten on exit.
1117N/A self.__update_repo_config()
1117N/A
1117N/A # Start the depot.
1117N/A self.dc.start()
1117N/A
1117N/A # Then, publish some packages we can abuse for testing.
1117N/A durl = self.dc.get_depot_url()
1895N/A plist = self.pkgsend_bulk(durl, (self.info10, self.quux10,
1895N/A self.system10, self.zfsextras10, self.zfsutils10))
1117N/A
1117N/A # Now, for each published package, attempt to get a p5i file
1117N/A # and then verify that the parsed response has the expected
1117N/A # package information under the expected publisher.
1117N/A for p in plist:
3234N/A purl = urljoin(durl, "p5i/0/{0}".format(p))
1117N/A pub, pkglist = p5i.parse(location=purl)[0]
1431N/A
1431N/A # p5i files contain non-qualified FMRIs as the FMRIs
1431N/A # are already grouped by publisher.
1431N/A nq_p = fmri.PkgFmri(p).get_fmri(anarchy=True,
1431N/A include_scheme=False)
1431N/A self.assertEqual(pkglist, [nq_p])
1117N/A
1117N/A # Try again, but only using package stems.
1117N/A for p in plist:
1431N/A stem = fmri.PkgFmri(p).pkg_name
3234N/A purl = urljoin(durl, "p5i/0/{0}".format(stem))
1117N/A pub, pkglist = p5i.parse(location=purl)[0]
1117N/A self.assertEqual(pkglist, [stem])
1117N/A
1117N/A # Try again, but using wildcards (which will return a list of
1117N/A # matching package stems).
3234N/A purl = urljoin(durl, "p5i/0/zfs*")
1117N/A pub, pkglist = p5i.parse(location=purl)[0]
1117N/A self.assertEqual(pkglist, ["zfs-extras", "zfs/utils"])
1117N/A
1117N/A # Finally, verify that a non-existent package will error out
1117N/A # with a httplib.NOT_FOUND.
1117N/A try:
3234N/A urlopen(urljoin(durl,
1117N/A "p5i/0/nosuchpackage"))
3234N/A except HTTPError as e:
3234N/A if e.code != http_client.NOT_FOUND:
1117N/A raise
1117N/A
1899N/A def test_3_headers(self):
1899N/A """Ensure expected headers are present for client operations
1899N/A (excluding publication)."""
1899N/A
1899N/A # Now update the repository configuration while the depot is
1899N/A # stopped so changes won't be overwritten on exit.
1899N/A self.__update_repo_config()
1899N/A
1899N/A # Start the depot.
1899N/A self.dc.start()
1899N/A
1899N/A durl = self.dc.get_depot_url()
2028N/A pfmri = fmri.PkgFmri(self.pkgsend_bulk(durl, self.file10,
2028N/A refresh_index=True)[0], "5.11")
1899N/A
2065N/A # Wait for search indexing to complete.
2065N/A self.wait_repo(self.dc.get_repodir())
2065N/A
1899N/A def get_headers(req_path):
1899N/A try:
3234N/A rinfo = urlopen(urljoin(durl,
1899N/A req_path)).info()
3234N/A return list(rinfo.items())
3234N/A except HTTPError as e:
3234N/A return list(e.info().items())
3171N/A except Exception as e:
3158N/A raise RuntimeError("retrieval of {0} "
3158N/A "failed: {1}".format(req_path, str(e)))
1899N/A
3329N/A for req_path in ("publisher/0",
1899N/A 'search/1/False_2_None_None_%2Fvar%2Ffile',
3158N/A "versions/0", "manifest/0/{0}".format(pfmri.get_url_path()),
3329N/A "catalog/1/catalog.attrs",
3274N/A "file/0/3aad0bca6f3a6f502c175700ebe90ef36e312d7e"):
1899N/A hdrs = dict(get_headers(req_path))
3339N/A if six.PY2:
3339N/A # Fields must be referenced in lowercase.
3339N/A cc = hdrs.get("cache-control", "")
3339N/A self.assertTrue(cc.startswith("must-revalidate, "
3339N/A "no-transform, max-age="))
3339N/A exp = hdrs.get("expires", None)
3339N/A else:
3339N/A # Fields begin with uppercase.
3339N/A cc = hdrs.get("Cache-Control", "")
3339N/A self.assertTrue(cc.startswith("must-revalidate, "
3339N/A "no-transform, max-age="))
3339N/A exp = hdrs.get("Expires", None)
3274N/A self.assertNotEqual(exp, None)
3274N/A self.assertTrue(exp.endswith(" GMT"))
1899N/A
2045N/A for req_path in ("catalog/1/catalog.hatters",
2045N/A "file/0/3aad0bca6f3a6f502c175700ebe90ef36e312d7f"):
2045N/A
2045N/A hdrs = dict(get_headers(req_path))
3339N/A if six.PY2:
3339N/A cc = hdrs.get("cache-control", None)
3339N/A prg = hdrs.get("pragma", None)
3339N/A else:
3339N/A cc = hdrs.get("Cache-Control", None)
3339N/A prg = hdrs.get("Pragma", None)
2045N/A self.assertEqual(cc, None)
2045N/A self.assertEqual(prg, None)
3130N/A
1850N/A def test_bug_15482(self):
1850N/A """Test to make sure BUI search doesn't trigger a traceback."""
1850N/A
1850N/A # Now update the repository configuration while the depot is
1850N/A # stopped so changes won't be overwritten on exit.
1850N/A self.__update_repo_config()
1850N/A
1850N/A # Start the depot.
1850N/A self.dc.start()
1850N/A
1850N/A # Then, publish some packages we can abuse for testing.
1850N/A durl = self.dc.get_depot_url()
2073N/A self.pkgsend_bulk(durl, self.quux10, refresh_index=True)
1895N/A
3234N/A surl = urljoin(durl,
1850N/A "en/search.shtml?action=Search&token=*")
3234N/A urlopen(surl).read()
3234N/A surl = urljoin(durl,
1850N/A "en/advanced_search.shtml?action=Search&token=*")
3234N/A urlopen(surl).read()
3234N/A surl = urljoin(durl,
1850N/A "en/advanced_search.shtml?token=*&show=a&rpp=50&"
1850N/A "action=Advanced+Search")
3234N/A urlopen(surl).read()
1850N/A
2230N/A def test_address(self):
2230N/A """Verify that depot address can be set."""
2230N/A
2230N/A # Check that IPv6 address can be used.
2230N/A self.dc.set_address("::1")
2272N/A self.dc.set_port(self.next_free_port)
2230N/A self.dc.start()
3339N/A self.assertTrue(self.dc.is_alive())
3339N/A self.assertTrue(self.dc.is_alive())
3339N/A self.assertTrue(self.dc.is_alive())
2230N/A
2230N/A # Check that we can retrieve something.
2230N/A durl = self.dc.get_depot_url()
3234N/A verdata = urlopen("{0}/versions/0/".format(durl))
2230N/A
2992N/A def test_log_depot_daemon(self):
2992N/A """Verify that depot daemon works properly and the error
2992N/A message is logged properly."""
2992N/A
2992N/A if self.dc.started:
2992N/A self.dc.stop()
2992N/A
2992N/A # Create a test repo.
2995N/A repopath = os.path.join(self.test_root, "repo_tmp_depot_log")
2992N/A self.create_repo(repopath)
2992N/A
2992N/A out_path = os.path.join(self.test_root, "daemon_out_log")
2992N/A err_path = os.path.join(self.test_root, "daemon_err_log")
2992N/A
2992N/A depot_handle = None
2992N/A try:
2992N/A depot_handle, curport = self.__depot_daemon_start(
2992N/A repopath, out_path, err_path)
2992N/A
2992N/A # Issue a bad request to trigger the server logging
2992N/A # the error msg.
3329N/A durl = "http://localhost:{0}/catalog/1/404".format(
3329N/A curport)
2992N/A try:
3234N/A urlopen(durl)
3234N/A except URLError as e:
2992N/A pass
2992N/A # Stop the depot daemon.
2992N/A self.__depot_daemon_stop(depot_handle)
2992N/A
2992N/A # Read the msgs from the err log file to verify log
2992N/A # msgs.
2992N/A with open(err_path, "r") as read_err:
2992N/A msgs = read_err.readlines()
3329N/A self.debug(durl)
3329N/A self.debug("".join(msgs))
2992N/A self.assertRegexp(msgs[-1], "\[[\w:/]+\] Request "
2992N/A "failed")
2992N/A except:
2992N/A self.__depot_daemon_stop(depot_handle)
2992N/A raise
2992N/A
817N/A
581N/Aif __name__ == "__main__":
581N/A unittest.main()