ff5a7e6377
https://fedoraproject.org/wiki/Changes/RemovePythonMockUsage Prefer using unittest.mock to a standalone package. The separate packages should only really be needed on Python 2.7 these days. The test requirements file is updated to only require mock on old Python, and the dependency is removed from setup.py to avoid issues there. Relates: https://src.fedoraproject.org/rpms/pungi/pull-request/9 Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
1087 lines
40 KiB
Python
1087 lines
40 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import argparse
|
|
|
|
try:
|
|
from unittest import mock
|
|
except ImportError:
|
|
import mock
|
|
import os
|
|
|
|
try:
|
|
import unittest2 as unittest
|
|
except ImportError:
|
|
import unittest
|
|
import tempfile
|
|
import shutil
|
|
import subprocess
|
|
import six
|
|
|
|
from pungi import compose
|
|
from pungi import util
|
|
|
|
from tests.helpers import touch, PungiTestCase, mk_boom, GIT_WITH_CREDS
|
|
|
|
|
|
class TestGitRefResolver(unittest.TestCase):
|
|
@mock.patch("pungi.util.run")
|
|
def test_successful_resolve(self, run):
|
|
run.return_value = (0, "CAFEBABE\tHEAD\n")
|
|
|
|
url = util.resolve_git_url("https://git.example.com/repo.git?somedir#HEAD")
|
|
|
|
self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE")
|
|
run.assert_called_once_with(
|
|
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
|
|
universal_newlines=True,
|
|
)
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_successful_resolve_with_credentials(self, run):
|
|
run.return_value = (0, "CAFEBABE\tHEAD\n")
|
|
|
|
url = util.resolve_git_url(
|
|
"https://git.example.com/repo.git?somedir#HEAD", "!ch"
|
|
)
|
|
|
|
self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE")
|
|
run.assert_called_once_with(
|
|
GIT_WITH_CREDS + ["ls-remote", "https://git.example.com/repo.git", "HEAD"],
|
|
universal_newlines=True,
|
|
)
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_successful_resolve_branch(self, run):
|
|
run.return_value = (0, "CAFEBABE\trefs/heads/f24\n")
|
|
|
|
url = util.resolve_git_url(
|
|
"https://git.example.com/repo.git?somedir#origin/f24"
|
|
)
|
|
|
|
self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE")
|
|
run.assert_called_once_with(
|
|
["git", "ls-remote", "https://git.example.com/repo.git", "refs/heads/f24"],
|
|
universal_newlines=True,
|
|
)
|
|
|
|
def test_resolve_ref_with_commit_id(self):
|
|
ref = util.resolve_git_ref("https://git.example.com/repo.git", "a" * 40)
|
|
self.assertEqual(ref, "a" * 40)
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_resolve_ref_multiple_matches(self, run):
|
|
run.return_value = (
|
|
0,
|
|
"CAFEBABE\trefs/heads/master\nBABECAFE\trefs/remotes/origin/master",
|
|
)
|
|
|
|
ref = util.resolve_git_ref("https://git.example.com/repo.git", "master")
|
|
|
|
self.assertEqual(ref, "CAFEBABE")
|
|
run.assert_called_once_with(
|
|
["git", "ls-remote", "https://git.example.com/repo.git", "master"],
|
|
universal_newlines=True,
|
|
)
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_resolve_ref_with_remote_head(self, run):
|
|
run.return_value = (0, "CAFEBABE\tHEAD\nBABECAFE\trefs/remotes/origin/HEAD")
|
|
|
|
ref = util.resolve_git_ref("https://git.example.com/repo.git", "HEAD")
|
|
|
|
self.assertEqual(ref, "CAFEBABE")
|
|
run.assert_called_once_with(
|
|
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
|
|
universal_newlines=True,
|
|
)
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_resolve_missing_spec(self, run):
|
|
url = util.resolve_git_url("https://git.example.com/repo.git")
|
|
|
|
self.assertEqual(url, "https://git.example.com/repo.git")
|
|
self.assertEqual(run.mock_calls, [])
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_resolve_non_head_spec(self, run):
|
|
url = util.resolve_git_url("https://git.example.com/repo.git#some-tag")
|
|
|
|
self.assertEqual(url, "https://git.example.com/repo.git#some-tag")
|
|
self.assertEqual(run.mock_calls, [])
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_resolve_ambiguous(self, run):
|
|
run.return_value = (0, "CAFEBABE\tF11\nDEADBEEF\tF10\n")
|
|
|
|
with self.assertRaises(RuntimeError):
|
|
util.resolve_git_url("https://git.example.com/repo.git?somedir#HEAD")
|
|
|
|
run.assert_called_once_with(
|
|
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
|
|
universal_newlines=True,
|
|
)
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_resolve_keep_empty_query_string(self, run):
|
|
run.return_value = (0, "CAFEBABE\tHEAD\n")
|
|
|
|
url = util.resolve_git_url("https://git.example.com/repo.git?#HEAD")
|
|
|
|
run.assert_called_once_with(
|
|
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
|
|
universal_newlines=True,
|
|
)
|
|
self.assertEqual(url, "https://git.example.com/repo.git?#CAFEBABE")
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_resolve_strip_git_plus_prefix(self, run):
|
|
run.return_value = (0, "CAFEBABE\tHEAD\n")
|
|
|
|
url = util.resolve_git_url("git+https://git.example.com/repo.git#HEAD")
|
|
|
|
run.assert_called_once_with(
|
|
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
|
|
universal_newlines=True,
|
|
)
|
|
self.assertEqual(url, "git+https://git.example.com/repo.git#CAFEBABE")
|
|
|
|
@mock.patch("pungi.util.run")
|
|
def test_resolve_no_branch_in_remote(self, run):
|
|
run.return_value = (0, "")
|
|
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
util.resolve_git_url(
|
|
"https://git.example.com/repo.git?somedir#origin/my-branch"
|
|
)
|
|
|
|
run.assert_called_once_with(
|
|
[
|
|
"git",
|
|
"ls-remote",
|
|
"https://git.example.com/repo.git",
|
|
"refs/heads/my-branch",
|
|
],
|
|
universal_newlines=True,
|
|
)
|
|
self.assertIn("ref does not exist in remote repo", str(ctx.exception))
|
|
|
|
@mock.patch("time.sleep")
|
|
@mock.patch("pungi.util.run")
|
|
def test_retry(self, run, sleep):
|
|
run.side_effect = [RuntimeError("Boom"), (0, "CAFEBABE\tHEAD\n")]
|
|
|
|
url = util.resolve_git_url("https://git.example.com/repo.git?somedir#HEAD")
|
|
|
|
self.assertEqual(url, "https://git.example.com/repo.git?somedir#CAFEBABE")
|
|
self.assertEqual(sleep.call_args_list, [mock.call(30)])
|
|
self.assertEqual(
|
|
run.call_args_list,
|
|
[
|
|
mock.call(
|
|
["git", "ls-remote", "https://git.example.com/repo.git", "HEAD"],
|
|
universal_newlines=True,
|
|
)
|
|
]
|
|
* 2,
|
|
)
|
|
|
|
@mock.patch("pungi.util.resolve_git_ref")
|
|
@mock.patch("pungi.util.resolve_git_url")
|
|
def test_resolver_offline(self, mock_resolve_url, mock_resolve_ref):
|
|
resolver = util.GitUrlResolver(offline=True)
|
|
self.assertEqual(
|
|
resolver("http://example.com/repo.git#HEAD"),
|
|
"http://example.com/repo.git#HEAD",
|
|
)
|
|
self.assertEqual(mock_resolve_url.call_args_list, [])
|
|
self.assertEqual(mock_resolve_ref.call_args_list, [])
|
|
|
|
@mock.patch("pungi.util.resolve_git_ref")
|
|
@mock.patch("pungi.util.resolve_git_url")
|
|
def test_resolver_offline_branch(self, mock_resolve_url, mock_resolve_ref):
|
|
resolver = util.GitUrlResolver(offline=True)
|
|
self.assertEqual(
|
|
resolver("http://example.com/repo.git", "master"),
|
|
"master",
|
|
)
|
|
self.assertEqual(mock_resolve_url.call_args_list, [])
|
|
self.assertEqual(mock_resolve_ref.call_args_list, [])
|
|
|
|
@mock.patch("pungi.util.resolve_git_ref")
|
|
@mock.patch("pungi.util.resolve_git_url")
|
|
def test_resolver_caches_calls(self, mock_resolve_url, mock_resolve_ref):
|
|
url1 = "http://example.com/repo.git#HEAD"
|
|
url2 = "http://example.com/repo.git#master"
|
|
url3 = "http://example.com/repo.git"
|
|
ref1 = "foo"
|
|
ref2 = "bar"
|
|
mock_resolve_url.side_effect = ["1", "2"]
|
|
mock_resolve_ref.side_effect = ["cafe", "beef"]
|
|
resolver = util.GitUrlResolver()
|
|
self.assertEqual(resolver(url1), "1")
|
|
self.assertEqual(resolver(url1), "1")
|
|
self.assertEqual(resolver(url3, ref1), "cafe")
|
|
self.assertEqual(resolver(url3, ref2), "beef")
|
|
self.assertEqual(resolver(url2), "2")
|
|
self.assertEqual(resolver(url3, ref1), "cafe")
|
|
self.assertEqual(resolver(url1), "1")
|
|
self.assertEqual(resolver(url3, ref2), "beef")
|
|
self.assertEqual(resolver(url2), "2")
|
|
self.assertEqual(resolver(url3, ref2), "beef")
|
|
self.assertEqual(
|
|
mock_resolve_url.call_args_list,
|
|
[mock.call(url1, None), mock.call(url2, None)],
|
|
)
|
|
self.assertEqual(
|
|
mock_resolve_ref.call_args_list,
|
|
[mock.call(url3, ref1, None), mock.call(url3, ref2, None)],
|
|
)
|
|
|
|
@mock.patch("pungi.util.resolve_git_url")
|
|
def test_resolver_caches_failure(self, mock_resolve):
|
|
url = "http://example.com/repo.git#HEAD"
|
|
mock_resolve.side_effect = mk_boom(util.GitUrlResolveError, "failed")
|
|
resolver = util.GitUrlResolver()
|
|
with self.assertRaises(util.GitUrlResolveError):
|
|
resolver(url)
|
|
with self.assertRaises(util.GitUrlResolveError):
|
|
resolver(url)
|
|
self.assertEqual(mock_resolve.call_args_list, [mock.call(url, None)])
|
|
|
|
|
|
class TestGetVariantData(unittest.TestCase):
|
|
def test_get_simple(self):
|
|
conf = {"foo": {"^Client$": 1}}
|
|
result = util.get_variant_data(conf, "foo", mock.Mock(uid="Client"))
|
|
self.assertEqual(result, [1])
|
|
|
|
def test_get_make_list(self):
|
|
conf = {"foo": {"^Client$": [1, 2], "^.*$": 3}}
|
|
result = util.get_variant_data(conf, "foo", mock.Mock(uid="Client"))
|
|
six.assertCountEqual(self, result, [1, 2, 3])
|
|
|
|
def test_not_matching_arch(self):
|
|
conf = {"foo": {"^Client$": [1, 2]}}
|
|
result = util.get_variant_data(conf, "foo", mock.Mock(uid="Server"))
|
|
self.assertEqual(result, [])
|
|
|
|
def test_handle_missing_config(self):
|
|
result = util.get_variant_data({}, "foo", mock.Mock(uid="Client"))
|
|
self.assertEqual(result, [])
|
|
|
|
def test_get_save_pattern(self):
|
|
conf = {"foo": {"^Client$": 1, "^NotClient$": 2}}
|
|
patterns = set()
|
|
result = util.get_variant_data(
|
|
conf, "foo", mock.Mock(uid="Client"), keys=patterns
|
|
)
|
|
self.assertEqual(result, [1])
|
|
self.assertEqual(patterns, set(["^Client$"]))
|
|
|
|
|
|
class TestVolumeIdGenerator(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp_dir)
|
|
|
|
@mock.patch("pungi.compose.ComposeInfo")
|
|
def test_get_volid(self, ci):
|
|
all_keys = [
|
|
(
|
|
["arch", "compose_id", "date", "disc_type"],
|
|
"x86_64-compose_id-20160107-",
|
|
),
|
|
(
|
|
["label", "label_major_version", "release_short", "respin"],
|
|
"RC-1.0-1-rel_short2-2",
|
|
),
|
|
(["type", "type_suffix", "variant", "version"], "nightly-.n-Server-6.0"),
|
|
]
|
|
for keys, expected in all_keys:
|
|
format = "-".join(["%(" + k + ")s" for k in keys])
|
|
conf = {
|
|
"release_short": "rel_short2",
|
|
"release_version": "6.0",
|
|
"image_volid_formats": [format],
|
|
"image_volid_layered_product_formats": [],
|
|
"volume_id_substitutions": {},
|
|
"restricted_volid": False,
|
|
}
|
|
variant = mock.Mock(uid="Server", type="variant")
|
|
ci.return_value.compose.respin = 2
|
|
ci.return_value.compose.id = "compose_id"
|
|
ci.return_value.compose.date = "20160107"
|
|
ci.return_value.compose.type = "nightly"
|
|
ci.return_value.compose.type_suffix = ".n"
|
|
ci.return_value.compose.label = "RC-1.0"
|
|
ci.return_value.compose.label_major_version = "1"
|
|
|
|
ci.return_value.release.version = "3.0"
|
|
ci.return_value.release.short = "rel_short"
|
|
|
|
c = compose.Compose(conf, self.tmp_dir)
|
|
|
|
volid = util.get_volid(c, "x86_64", variant, disc_type=False)
|
|
|
|
self.assertEqual(volid, expected)
|
|
|
|
@mock.patch("pungi.compose.ComposeInfo")
|
|
def test_get_restricted_volid(self, ci):
|
|
all_keys = [
|
|
(
|
|
["arch", "compose_id", "date", "disc_type"],
|
|
"x86_64-compose_id-20160107-",
|
|
),
|
|
(
|
|
["label", "label_major_version", "release_short", "respin"],
|
|
"RC-1-0-1-rel_short2-2",
|
|
),
|
|
(["type", "type_suffix", "variant", "version"], "nightly--n-Server-6-0"),
|
|
]
|
|
for keys, expected in all_keys:
|
|
format = "-".join(["%(" + k + ")s" for k in keys])
|
|
conf = {
|
|
"release_short": "rel_short2",
|
|
"release_version": "6.0",
|
|
"image_volid_formats": [format],
|
|
"image_volid_layered_product_formats": [],
|
|
"volume_id_substitutions": {},
|
|
"restricted_volid": True,
|
|
}
|
|
variant = mock.Mock(uid="Server", type="variant")
|
|
ci.return_value.compose.respin = 2
|
|
ci.return_value.compose.id = "compose_id"
|
|
ci.return_value.compose.date = "20160107"
|
|
ci.return_value.compose.type = "nightly"
|
|
ci.return_value.compose.type_suffix = ".n"
|
|
ci.return_value.compose.label = "RC-1.0"
|
|
ci.return_value.compose.label_major_version = "1"
|
|
|
|
ci.return_value.release.version = "3.0"
|
|
ci.return_value.release.short = "rel_short"
|
|
|
|
c = compose.Compose(conf, self.tmp_dir)
|
|
|
|
volid = util.get_volid(c, "x86_64", variant, disc_type=False)
|
|
|
|
self.assertEqual(volid, expected)
|
|
|
|
@mock.patch("pungi.compose.ComposeInfo")
|
|
def test_get_volid_too_long(self, ci):
|
|
conf = {
|
|
"release_short": "rel_short2",
|
|
"release_version": "6.0",
|
|
"image_volid_formats": [
|
|
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", # 34 chars
|
|
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", # 33 chars
|
|
],
|
|
"image_volid_layered_product_formats": [],
|
|
"volume_id_substitutions": {},
|
|
}
|
|
variant = mock.Mock(uid="Server", type="variant")
|
|
c = compose.Compose(conf, self.tmp_dir)
|
|
|
|
with self.assertRaises(ValueError) as ctx:
|
|
util.get_volid(c, "x86_64", variant, disc_type=False)
|
|
|
|
self.assertIn("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", str(ctx.exception))
|
|
self.assertIn("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", str(ctx.exception))
|
|
|
|
@mock.patch("pungi.compose.ComposeInfo")
|
|
def test_apply_substitutions(self, ci):
|
|
all_keys = [
|
|
(
|
|
"Fedora-WorkstationOstree-ostree-x86_64-rawhide",
|
|
"Fedora-WS-ostree-x86_64-rawhide",
|
|
),
|
|
(
|
|
"Fedora-WorkstationOstree-ostree-x86_64-Rawhide",
|
|
"Fedora-WS-ostree-x86_64-rawh",
|
|
),
|
|
("x86_64-compose_id-20160107", "x86_64-compose_id-20160107"),
|
|
("x86_64-compose_id-20160107-Alpha", "x86_64-compose_id-20160107-A"),
|
|
# These test the case where one substitution is a subset
|
|
# of the other, but sorts alphabetically ahead of it, to
|
|
# make sure we're correctly sorting by length
|
|
("Fedora-zzzaaaaaazzz-Rawhide", "Fedora-zzz-rawh"),
|
|
("Fedora-aaaaaa-Rawhide", "Fedora-aaa-rawh"),
|
|
]
|
|
for volid, expected in all_keys:
|
|
conf = {
|
|
"volume_id_substitutions": {
|
|
"Rawhide": "rawh",
|
|
"WorkstationOstree": "WS",
|
|
"Workstation": "WS",
|
|
"Alpha": "A",
|
|
"zzzaaaaaazzz": "zzz",
|
|
"aaaaaa": "aaa",
|
|
}
|
|
}
|
|
c = compose.Compose(conf, self.tmp_dir)
|
|
transformed_volid = util._apply_substitutions(c, volid)
|
|
self.assertEqual(transformed_volid, expected)
|
|
|
|
|
|
class TestFindOldCompose(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp_dir)
|
|
|
|
def test_finds_single(self):
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160229.0/STATUS", "FINISHED")
|
|
old = util.find_old_compose(self.tmp_dir, "Fedora", "Rawhide", "")
|
|
self.assertEqual(old, self.tmp_dir + "/Fedora-Rawhide-20160229.0")
|
|
|
|
def test_ignores_in_progress(self):
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160229.0/STATUS", "STARTED")
|
|
old = util.find_old_compose(self.tmp_dir, "Fedora", "Rawhide", "")
|
|
self.assertIsNone(old)
|
|
|
|
def test_only_considers_allowed_status(self):
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160229.0/STATUS", "FINISHED")
|
|
old = util.find_old_compose(
|
|
self.tmp_dir, "Fedora", "Rawhide", "", allowed_statuses=["DOOMED"]
|
|
)
|
|
self.assertIsNone(old)
|
|
|
|
def test_finds_latest(self):
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160228.0/STATUS", "DOOMED")
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160229.0/STATUS", "FINISHED")
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160229.1/STATUS", "FINISHED_INCOMPLETE")
|
|
old = util.find_old_compose(self.tmp_dir, "Fedora", "Rawhide", "")
|
|
self.assertEqual(old, self.tmp_dir + "/Fedora-Rawhide-20160229.1")
|
|
|
|
def test_find_correct_type(self):
|
|
touch(self.tmp_dir + "/Fedora-26-updates-20160229.0/STATUS", "FINISHED")
|
|
touch(self.tmp_dir + "/Fedora-26-updates-testing-20160229.0/STATUS", "FINISHED")
|
|
old = util.find_old_compose(self.tmp_dir, "Fedora", "26", "-updates")
|
|
self.assertEqual(old, self.tmp_dir + "/Fedora-26-updates-20160229.0")
|
|
old = util.find_old_compose(self.tmp_dir, "Fedora", "26", "-updates-testing")
|
|
self.assertEqual(old, self.tmp_dir + "/Fedora-26-updates-testing-20160229.0")
|
|
|
|
def test_find_latest_with_two_digit_respin(self):
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160228.n.9/STATUS", "FINISHED")
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160228.n.10/STATUS", "FINISHED")
|
|
old = util.find_old_compose(self.tmp_dir, "Fedora", "Rawhide", "")
|
|
self.assertEqual(old, self.tmp_dir + "/Fedora-Rawhide-20160228.n.10")
|
|
|
|
def test_finds_ignores_other_files(self):
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160229.0", "not a compose")
|
|
touch(
|
|
self.tmp_dir + "/Fedora-Rawhide-20160228.0/STATUS/file",
|
|
"also not a compose",
|
|
)
|
|
touch(self.tmp_dir + "/Fedora-24-20160229.0/STATUS", "FINISHED")
|
|
touch(self.tmp_dir + "/Another-Rawhide-20160229.0/STATUS", "FINISHED")
|
|
old = util.find_old_compose(self.tmp_dir, "Fedora", "Rawhide", "")
|
|
self.assertIsNone(old)
|
|
|
|
def test_search_in_file(self):
|
|
touch(self.tmp_dir + "/file")
|
|
old = util.find_old_compose(self.tmp_dir + "/file", "Fedora", "Rawhide", "")
|
|
self.assertIsNone(old)
|
|
|
|
def test_do_not_skip_symlink(self):
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-20160228.n.10/STATUS", "FINISHED")
|
|
os.symlink(
|
|
self.tmp_dir + "/Fedora-Rawhide-20160228.n.10",
|
|
self.tmp_dir + "/Fedora-Rawhide-20160229.n.0",
|
|
)
|
|
old = util.find_old_compose(self.tmp_dir, "Fedora", "Rawhide", "")
|
|
self.assertEqual(old, self.tmp_dir + "/Fedora-Rawhide-20160229.n.0")
|
|
|
|
def test_finds_layered_product(self):
|
|
touch(self.tmp_dir + "/Fedora-Rawhide-Base-1-20160229.0/STATUS", "FINISHED")
|
|
old = util.find_old_compose(
|
|
self.tmp_dir,
|
|
"Fedora",
|
|
"Rawhide",
|
|
"",
|
|
base_product_short="Base",
|
|
base_product_version="1",
|
|
)
|
|
self.assertEqual(old, self.tmp_dir + "/Fedora-Rawhide-Base-1-20160229.0")
|
|
|
|
|
|
class TestHelpers(PungiTestCase):
|
|
def test_process_args(self):
|
|
self.assertEqual(util.process_args("--opt=%s", None), [])
|
|
self.assertEqual(util.process_args("--opt=%s", []), [])
|
|
self.assertEqual(
|
|
util.process_args("--opt=%s", ["foo", "bar"]), ["--opt=foo", "--opt=bar"]
|
|
)
|
|
self.assertEqual(util.process_args("--opt=%s", "foo"), ["--opt=foo"])
|
|
|
|
def test_makedirs(self):
|
|
util.makedirs(self.topdir + "/foo/bar/baz")
|
|
self.assertTrue(os.path.isdir(self.topdir + "/foo/bar/baz"))
|
|
|
|
def test_makedirs_on_existing(self):
|
|
os.makedirs(self.topdir + "/foo/bar/baz")
|
|
try:
|
|
util.makedirs(self.topdir + "/foo/bar/baz")
|
|
except OSError:
|
|
self.fail("makedirs raised exception on existing directory")
|
|
|
|
|
|
class TestLevenshtein(unittest.TestCase):
|
|
def test_edit_dist_empty_str(self):
|
|
self.assertEqual(util.levenshtein("", ""), 0)
|
|
|
|
def test_edit_dist_same_str(self):
|
|
self.assertEqual(util.levenshtein("aaa", "aaa"), 0)
|
|
|
|
def test_edit_dist_one_change(self):
|
|
self.assertEqual(util.levenshtein("aab", "aaa"), 1)
|
|
|
|
def test_edit_dist_different_words(self):
|
|
self.assertEqual(util.levenshtein("kitten", "sitting"), 3)
|
|
|
|
|
|
class TestRecursiveFileList(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp_dir)
|
|
|
|
def test_flat_file_list(self):
|
|
"""Build a directory containing files and assert they are listed."""
|
|
expected_files = sorted(["file1", "file2", "file3"])
|
|
for expected_file in [os.path.join(self.tmp_dir, f) for f in expected_files]:
|
|
touch(expected_file)
|
|
|
|
actual_files = sorted(util.recursive_file_list(self.tmp_dir))
|
|
self.assertEqual(expected_files, actual_files)
|
|
|
|
def test_nested_file_list(self):
|
|
"""Build a directory containing files and assert they are listed."""
|
|
expected_files = sorted(["file1", "subdir/file2", "sub/subdir/file3"])
|
|
for expected_file in [os.path.join(self.tmp_dir, f) for f in expected_files]:
|
|
touch(expected_file)
|
|
|
|
actual_files = sorted(util.recursive_file_list(self.tmp_dir))
|
|
self.assertEqual(expected_files, actual_files)
|
|
|
|
|
|
class TestTempFiles(unittest.TestCase):
|
|
def test_temp_dir_ok(self):
|
|
with util.temp_dir() as tmp:
|
|
self.assertTrue(os.path.isdir(tmp))
|
|
self.assertFalse(os.path.exists(tmp))
|
|
|
|
def test_temp_dir_fail(self):
|
|
with self.assertRaises(RuntimeError):
|
|
with util.temp_dir() as tmp:
|
|
self.assertTrue(os.path.isdir(tmp))
|
|
raise RuntimeError("BOOM")
|
|
self.assertFalse(os.path.exists(tmp))
|
|
|
|
def test_temp_dir_in_non_existing_dir(self):
|
|
with util.temp_dir() as playground:
|
|
root = os.path.join(playground, "missing")
|
|
with util.temp_dir(dir=root) as tmp:
|
|
self.assertTrue(os.path.isdir(tmp))
|
|
self.assertTrue(os.path.isdir(root))
|
|
self.assertFalse(os.path.exists(tmp))
|
|
|
|
|
|
class TestUnmountCmd(unittest.TestCase):
|
|
def _fakeProc(self, ret, err="", out=""):
|
|
proc = mock.Mock(returncode=ret)
|
|
proc.communicate.return_value = (out, err)
|
|
return proc
|
|
|
|
@mock.patch("subprocess.Popen")
|
|
def test_unmount_cmd_success(self, mockPopen):
|
|
cmd = "unmount"
|
|
mockPopen.side_effect = [self._fakeProc(0, "")]
|
|
util.run_unmount_cmd(cmd)
|
|
self.assertEqual(
|
|
mockPopen.call_args_list,
|
|
[
|
|
mock.call(
|
|
cmd,
|
|
stderr=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
)
|
|
],
|
|
)
|
|
|
|
@mock.patch("subprocess.Popen")
|
|
def test_unmount_cmd_fail_other_reason(self, mockPopen):
|
|
cmd = "unmount"
|
|
mockPopen.side_effect = [self._fakeProc(1, "It is broken")]
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
util.run_unmount_cmd(cmd)
|
|
self.assertEqual(
|
|
str(ctx.exception), "Unhandled error when running 'unmount': 'It is broken'"
|
|
)
|
|
self.assertEqual(
|
|
mockPopen.call_args_list,
|
|
[
|
|
mock.call(
|
|
cmd,
|
|
stderr=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
)
|
|
],
|
|
)
|
|
|
|
@mock.patch("time.sleep")
|
|
@mock.patch("subprocess.Popen")
|
|
def test_unmount_cmd_fail_then_retry(self, mockPopen, mock_sleep):
|
|
cmd = "unmount"
|
|
mockPopen.side_effect = [
|
|
self._fakeProc(1, "Device or resource busy"),
|
|
self._fakeProc(1, "Device or resource busy"),
|
|
self._fakeProc(0, ""),
|
|
]
|
|
util.run_unmount_cmd(cmd)
|
|
self.assertEqual(
|
|
mockPopen.call_args_list,
|
|
[
|
|
mock.call(
|
|
cmd,
|
|
stderr=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
)
|
|
]
|
|
* 3,
|
|
)
|
|
self.assertEqual(mock_sleep.call_args_list, [mock.call(0), mock.call(1)])
|
|
|
|
@mock.patch("time.sleep")
|
|
@mock.patch("subprocess.Popen")
|
|
def test_unmount_cmd_fail_then_retry_and_fail(self, mockPopen, mock_sleep):
|
|
cmd = "unmount"
|
|
mockPopen.side_effect = [
|
|
self._fakeProc(1, "Device or resource busy"),
|
|
self._fakeProc(1, "Device or resource busy"),
|
|
self._fakeProc(1, "Device or resource busy"),
|
|
]
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
util.run_unmount_cmd(cmd, max_retries=3)
|
|
self.assertEqual(
|
|
mockPopen.call_args_list,
|
|
[
|
|
mock.call(
|
|
cmd,
|
|
stderr=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
)
|
|
]
|
|
* 3,
|
|
)
|
|
self.assertEqual(
|
|
mock_sleep.call_args_list, [mock.call(0), mock.call(1), mock.call(2)]
|
|
)
|
|
self.assertEqual(
|
|
str(ctx.exception), "Failed to run 'unmount': Device or resource busy."
|
|
)
|
|
|
|
@mock.patch("time.sleep")
|
|
@mock.patch("subprocess.Popen")
|
|
def test_fusermount_fail_then_retry_and_fail_with_debug(
|
|
self, mockPopen, mock_sleep
|
|
):
|
|
logger = mock.Mock()
|
|
mockPopen.side_effect = [
|
|
self._fakeProc(1, "Device or resource busy"),
|
|
self._fakeProc(1, "Device or resource busy"),
|
|
self._fakeProc(1, "Device or resource busy"),
|
|
self._fakeProc(0, out="list of files"),
|
|
self._fakeProc(0, out="It is very busy"),
|
|
self._fakeProc(1, out="lsof output"),
|
|
]
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
util.run_unmount_cmd(
|
|
["fusermount", "-u", "/path"],
|
|
path="/path",
|
|
max_retries=3,
|
|
logger=logger,
|
|
)
|
|
cmd = ["fusermount", "-u", "/path"]
|
|
expected = [
|
|
mock.call(
|
|
cmd,
|
|
stderr=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
),
|
|
mock.call(
|
|
cmd,
|
|
stderr=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
),
|
|
mock.call(
|
|
cmd,
|
|
stderr=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
),
|
|
mock.call(
|
|
["ls", "-lA", "/path"],
|
|
stderr=subprocess.STDOUT,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
),
|
|
mock.call(
|
|
["fuser", "-vm", "/path"],
|
|
stderr=subprocess.STDOUT,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
),
|
|
mock.call(
|
|
["lsof", "+D", "/path"],
|
|
stderr=subprocess.STDOUT,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
),
|
|
]
|
|
self.assertEqual(mockPopen.call_args_list, expected)
|
|
self.assertEqual(
|
|
mock_sleep.call_args_list, [mock.call(0), mock.call(1), mock.call(2)]
|
|
)
|
|
self.assertEqual(
|
|
str(ctx.exception),
|
|
"Failed to run ['fusermount', '-u', '/path']: Device or resource busy.",
|
|
)
|
|
self.assertEqual(
|
|
logger.mock_calls,
|
|
[
|
|
mock.call.debug(
|
|
"`%s` exited with %s and following output:\n%s",
|
|
"ls -lA /path",
|
|
0,
|
|
"list of files",
|
|
),
|
|
mock.call.debug(
|
|
"`%s` exited with %s and following output:\n%s",
|
|
"fuser -vm /path",
|
|
0,
|
|
"It is very busy",
|
|
),
|
|
mock.call.debug(
|
|
"`%s` exited with %s and following output:\n%s",
|
|
"lsof +D /path",
|
|
1,
|
|
"lsof output",
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
class TranslatePathTestCase(unittest.TestCase):
|
|
def test_does_nothing_without_config(self):
|
|
compose = mock.Mock(conf={"translate_paths": []})
|
|
ret = util.translate_path(compose, "/mnt/koji/compose/rawhide/XYZ")
|
|
self.assertEqual(ret, "/mnt/koji/compose/rawhide/XYZ")
|
|
|
|
def test_translates_prefix(self):
|
|
compose = mock.Mock(
|
|
conf={"translate_paths": [("/mnt/koji", "http://example.com")]}
|
|
)
|
|
ret = util.translate_path(compose, "/mnt/koji/compose/rawhide/XYZ")
|
|
self.assertEqual(ret, "http://example.com/compose/rawhide/XYZ")
|
|
|
|
def test_does_not_translate_not_matching(self):
|
|
compose = mock.Mock(
|
|
conf={"translate_paths": [("/mnt/koji", "http://example.com")]}
|
|
)
|
|
ret = util.translate_path(compose, "/mnt/fedora_koji/compose/rawhide/XYZ")
|
|
self.assertEqual(ret, "/mnt/fedora_koji/compose/rawhide/XYZ")
|
|
|
|
|
|
class GetRepoFuncsTestCase(unittest.TestCase):
|
|
@mock.patch("pungi.compose.ComposeInfo")
|
|
def setUp(self, ci):
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
conf = {"translate_paths": [(self.tmp_dir, "http://example.com")]}
|
|
ci.return_value.compose.respin = 0
|
|
ci.return_value.compose.id = "RHEL-8.0-20180101.n.0"
|
|
ci.return_value.compose.date = "20160101"
|
|
ci.return_value.compose.type = "nightly"
|
|
ci.return_value.compose.type_suffix = ".n"
|
|
ci.return_value.compose.label = "RC-1.0"
|
|
ci.return_value.compose.label_major_version = "1"
|
|
|
|
compose_dir = os.path.join(self.tmp_dir, ci.return_value.compose.id)
|
|
self.compose = compose.Compose(conf, compose_dir)
|
|
server_variant = mock.Mock(uid="Server", type="variant")
|
|
client_variant = mock.Mock(uid="Client", type="variant")
|
|
self.compose.all_variants = {
|
|
"Server": server_variant,
|
|
"Client": client_variant,
|
|
}
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp_dir)
|
|
|
|
def test_get_repo_url_from_normal_url(self):
|
|
url = util.get_repo_url(self.compose, "http://example.com/repo")
|
|
self.assertEqual(url, "http://example.com/repo")
|
|
|
|
def test_get_repo_url_from_path(self):
|
|
url = util.get_repo_url(self.compose, os.path.join(self.tmp_dir, "repo"))
|
|
self.assertEqual(url, "http://example.com/repo")
|
|
|
|
def test_get_repo_url_from_variant_uid(self):
|
|
url = util.get_repo_url(self.compose, "Server")
|
|
self.assertEqual(
|
|
url, "http://example.com/RHEL-8.0-20180101.n.0/compose/Server/$basearch/os"
|
|
)
|
|
|
|
def test_get_repo_url_from_repo_dict(self):
|
|
repo = {"baseurl": "http://example.com/repo"}
|
|
url = util.get_repo_url(self.compose, repo)
|
|
self.assertEqual(url, "http://example.com/repo")
|
|
|
|
repo = {"baseurl": "Server"}
|
|
url = util.get_repo_url(self.compose, repo)
|
|
self.assertEqual(
|
|
url, "http://example.com/RHEL-8.0-20180101.n.0/compose/Server/$basearch/os"
|
|
)
|
|
|
|
def test_get_repo_urls(self):
|
|
repos = [
|
|
"http://example.com/repo",
|
|
"Server",
|
|
{"baseurl": "Client"},
|
|
{"baseurl": "ftp://example.com/linux/repo"},
|
|
]
|
|
|
|
expect = [
|
|
"http://example.com/repo",
|
|
"http://example.com/RHEL-8.0-20180101.n.0/compose/Server/$basearch/os",
|
|
"http://example.com/RHEL-8.0-20180101.n.0/compose/Client/$basearch/os",
|
|
"ftp://example.com/linux/repo",
|
|
]
|
|
|
|
self.assertEqual(util.get_repo_urls(self.compose, repos), expect)
|
|
|
|
def test_get_repo_dict_from_normal_url(self):
|
|
repo_dict = util.get_repo_dict("http://example.com/repo")
|
|
expect = {
|
|
"name": "http:__example.com_repo",
|
|
"baseurl": "http://example.com/repo",
|
|
}
|
|
self.assertEqual(repo_dict, expect)
|
|
|
|
def test_get_repo_dict_from_variant_uid(self):
|
|
repo_dict = util.get_repo_dict("Server") # this repo format is deprecated
|
|
expect = {}
|
|
self.assertEqual(repo_dict, expect)
|
|
|
|
def test_get_repo_dict_from_repo_dict(self):
|
|
repo = {"baseurl": "Server"} # this repo format is deprecated
|
|
expect = {}
|
|
repo_dict = util.get_repo_dict(repo)
|
|
self.assertEqual(repo_dict, expect)
|
|
|
|
def test_get_repo_dicts(self):
|
|
repos = [
|
|
"http://example.com/repo",
|
|
"Server", # this repo format is deprecated (and will not be included into final repo_dict) # noqa: E501
|
|
{"baseurl": "Client"}, # this repo format is deprecated
|
|
{"baseurl": "ftp://example.com/linux/repo"},
|
|
{"name": "testrepo", "baseurl": "ftp://example.com/linux/repo"},
|
|
]
|
|
expect = [
|
|
{"name": "http:__example.com_repo", "baseurl": "http://example.com/repo"},
|
|
{
|
|
"name": "ftp:__example.com_linux_repo",
|
|
"baseurl": "ftp://example.com/linux/repo",
|
|
},
|
|
{"name": "testrepo", "baseurl": "ftp://example.com/linux/repo"},
|
|
]
|
|
repos = util.get_repo_dicts(repos)
|
|
self.assertEqual(repos, expect)
|
|
|
|
|
|
class TestVersionGenerator(unittest.TestCase):
|
|
def setUp(self):
|
|
ci = mock.MagicMock()
|
|
ci.respin = 0
|
|
ci.id = "RHEL-8.0-20180101.0"
|
|
ci.release.version = "8"
|
|
ci.type = "nightly"
|
|
ci.type_suffix = ""
|
|
ci.label = "RC-1.0"
|
|
ci.label_major_version = "1"
|
|
|
|
self.compose = mock.MagicMock()
|
|
self.compose.ci_base = ci
|
|
self.compose.compose_respin = 0
|
|
self.compose.compose_date = "20160101"
|
|
|
|
def test_unknown_generator(self):
|
|
compose = mock.Mock()
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
util.version_generator(compose, "!GIMME_VERSION")
|
|
|
|
self.assertEqual(
|
|
str(ctx.exception), "Unknown version generator '!GIMME_VERSION'"
|
|
)
|
|
|
|
def test_passthrough_value(self):
|
|
compose = mock.Mock()
|
|
self.assertEqual(util.version_generator(compose, "1.2.3"), "1.2.3")
|
|
|
|
def test_passthrough_none(self):
|
|
compose = mock.Mock()
|
|
self.assertEqual(util.version_generator(compose, None), None)
|
|
|
|
def test_release_from_version_date_respin(self):
|
|
self.assertEqual(
|
|
util.version_generator(self.compose, "!VERSION_FROM_VERSION_DATE_RESPIN"),
|
|
"8.20160101.0",
|
|
)
|
|
|
|
def test_release_from_date_respin(self):
|
|
self.assertEqual(
|
|
util.version_generator(self.compose, "!RELEASE_FROM_DATE_RESPIN"),
|
|
"20160101.0",
|
|
)
|
|
|
|
def test_version_from_version(self):
|
|
self.assertEqual(
|
|
util.version_generator(self.compose, "!VERSION_FROM_VERSION"),
|
|
"8",
|
|
)
|
|
|
|
|
|
class TestTZOffset(unittest.TestCase):
|
|
@mock.patch("time.daylight", new=False)
|
|
@mock.patch("time.altzone", new=7200)
|
|
@mock.patch("time.timezone", new=3600)
|
|
@mock.patch("time.localtime", new=lambda: mock.Mock(tm_isdst=0))
|
|
def test_zone_without_dst(self):
|
|
self.assertEqual(util.get_tz_offset(), "-01:00")
|
|
|
|
@mock.patch("time.daylight", new=True)
|
|
@mock.patch("time.altzone", new=7200)
|
|
@mock.patch("time.timezone", new=3600)
|
|
@mock.patch("time.localtime", new=lambda: mock.Mock(tm_isdst=0))
|
|
def test_with_active_dst(self):
|
|
self.assertEqual(util.get_tz_offset(), "-01:00")
|
|
|
|
@mock.patch("time.daylight", new=True)
|
|
@mock.patch("time.altzone", new=-9000)
|
|
@mock.patch("time.timezone", new=-3600)
|
|
@mock.patch("time.localtime", new=lambda: mock.Mock(tm_isdst=1))
|
|
def test_with_inactive_dst(self):
|
|
self.assertEqual(util.get_tz_offset(), "+02:30")
|
|
|
|
@mock.patch("time.daylight", new=False)
|
|
@mock.patch("time.altzone", new=0)
|
|
@mock.patch("time.timezone", new=0)
|
|
@mock.patch("time.localtime", new=lambda: mock.Mock(tm_isdst=0))
|
|
def test_utc(self):
|
|
self.assertEqual(util.get_tz_offset(), "+00:00")
|
|
|
|
|
|
class TestParseKojiEvent(PungiTestCase):
|
|
def test_number(self):
|
|
self.assertEqual(util.parse_koji_event("1234"), 1234)
|
|
|
|
def test_correct_path(self):
|
|
touch(
|
|
os.path.join(self.topdir, "work/global/koji-event"),
|
|
'{"id": 19769058, "ts": 1527641311.22855}',
|
|
)
|
|
|
|
self.assertEqual(util.parse_koji_event(self.topdir), 19769058)
|
|
|
|
def test_bad_path(self):
|
|
with self.assertRaises(argparse.ArgumentTypeError):
|
|
util.parse_koji_event(self.topdir)
|
|
|
|
|
|
class TestCopyAll(PungiTestCase):
|
|
def setUp(self):
|
|
super(TestCopyAll, self).setUp()
|
|
self.src = os.path.join(self.topdir, "src")
|
|
self.dst = os.path.join(self.topdir, "dst")
|
|
util.makedirs(self.src)
|
|
|
|
def test_preserve_symlink(self):
|
|
touch(os.path.join(self.src, "target"))
|
|
os.symlink("target", os.path.join(self.src, "symlink"))
|
|
|
|
util.copy_all(self.src, self.dst)
|
|
|
|
self.assertTrue(os.path.isfile(os.path.join(self.dst, "target")))
|
|
self.assertTrue(os.path.islink(os.path.join(self.dst, "symlink")))
|
|
self.assertEqual(os.readlink(os.path.join(self.dst, "symlink")), "target")
|
|
|
|
def test_copy_broken_symlink(self):
|
|
os.symlink("broken", os.path.join(self.src, "symlink"))
|
|
|
|
util.copy_all(self.src, self.dst)
|
|
|
|
self.assertTrue(os.path.islink(os.path.join(self.dst, "symlink")))
|
|
self.assertEqual(os.readlink(os.path.join(self.dst, "symlink")), "broken")
|
|
|
|
|
|
class TestMoveAll(PungiTestCase):
|
|
def setUp(self):
|
|
super(TestMoveAll, self).setUp()
|
|
self.src = os.path.join(self.topdir, "src")
|
|
self.dst = os.path.join(self.topdir, "dst")
|
|
util.makedirs(self.src)
|
|
|
|
def test_move_all(self):
|
|
touch(os.path.join(self.src, "target"))
|
|
util.move_all(self.src, self.dst)
|
|
|
|
self.assertTrue(os.path.isfile(os.path.join(self.dst, "target")))
|
|
self.assertTrue(os.path.exists(os.path.join(self.src)))
|
|
self.assertFalse(os.path.isfile(os.path.join(self.src, "target")))
|
|
|
|
def test_move_all_rm_src_dir(self):
|
|
touch(os.path.join(self.src, "target"))
|
|
util.move_all(self.src, self.dst, rm_src_dir=True)
|
|
|
|
self.assertTrue(os.path.isfile(os.path.join(self.dst, "target")))
|
|
self.assertFalse(os.path.exists(os.path.join(self.src)))
|
|
self.assertFalse(os.path.isfile(os.path.join(self.src, "target")))
|
|
|
|
|
|
@mock.patch("six.moves.urllib.request.urlretrieve")
|
|
class TestAsLocalFile(PungiTestCase):
|
|
def test_local_file(self, urlretrieve):
|
|
with util.as_local_file("/tmp/foo") as fn:
|
|
self.assertEqual(fn, "/tmp/foo")
|
|
self.assertEqual(urlretrieve.call_args_list, [])
|
|
|
|
def test_http(self, urlretrieve):
|
|
url = "http://example.com/repodata/repomd.xml"
|
|
|
|
def my_mock(url_):
|
|
self.assertEqual(url, url_)
|
|
self.filename = os.path.join(self.topdir, "my-file")
|
|
touch(self.filename)
|
|
return self.filename, {}
|
|
|
|
urlretrieve.side_effect = my_mock
|
|
|
|
with util.as_local_file(url) as fn:
|
|
self.assertEqual(fn, self.filename)
|
|
self.assertTrue(os.path.exists(self.filename))
|
|
self.assertFalse(os.path.exists(self.filename))
|
|
|
|
def test_file_url(self, urlretrieve):
|
|
with util.as_local_file("file:///tmp/foo") as fn:
|
|
self.assertEqual(fn, "/tmp/foo")
|
|
self.assertEqual(urlretrieve.call_args_list, [])
|