aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorBrian Harring <ferringb@gmail.com>2022-12-25 11:26:31 -0800
committerArthur Zamarin <arthurzam@gentoo.org>2022-12-25 21:32:26 +0200
commitb17af120f9081aa30d6a368fd75a69c298cb70e0 (patch)
tree3de8bc1fba4fbd004b9b1703e67b1a9feb5ab6fc /tests/sync
parentFix secondary exception from when an EBD fails to start. (diff)
downloadpkgcore-b17af120f9081aa30d6a368fd75a69c298cb70e0.tar.gz
pkgcore-b17af120f9081aa30d6a368fd75a69c298cb70e0.tar.bz2
pkgcore-b17af120f9081aa30d6a368fd75a69c298cb70e0.zip
Reformat w/ black-22.12.0
Signed-off-by: Brian Harring <ferringb@gmail.com> Signed-off-by: Arthur Zamarin <arthurzam@gentoo.org>
Diffstat (limited to 'tests/sync')
-rw-r--r--tests/sync/test_base.py76
-rw-r--r--tests/sync/test_bzr.py11
-rw-r--r--tests/sync/test_cvs.py26
-rw-r--r--tests/sync/test_darcs.py11
-rw-r--r--tests/sync/test_git.py33
-rw-r--r--tests/sync/test_git_svn.py15
-rw-r--r--tests/sync/test_hg.py28
-rw-r--r--tests/sync/test_rsync.py56
-rw-r--r--tests/sync/test_sqfs.py13
-rw-r--r--tests/sync/test_svn.py11
-rw-r--r--tests/sync/test_tar.py11
11 files changed, 142 insertions, 149 deletions
diff --git a/tests/sync/test_base.py b/tests/sync/test_base.py
index 34511541..22fb6e82 100644
--- a/tests/sync/test_base.py
+++ b/tests/sync/test_base.py
@@ -12,10 +12,9 @@ existing_uid = pwd.getpwnam(existing_user).pw_uid
class TestSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = str(tmp_path / 'repo')
+ self.repo_path = str(tmp_path / "repo")
def test_split_users(self):
o = base.Syncer(self.repo_path, "http://dar")
@@ -37,23 +36,23 @@ class TestSyncer:
with pytest.raises(base.MissingLocalUser):
base.Syncer(self.repo_path, f"foo_nonexistent_user::foon@site")
- @mock.patch('snakeoil.process.spawn.spawn')
+ @mock.patch("snakeoil.process.spawn.spawn")
def test_usersync_disabled(self, spawn):
o = base.Syncer(self.repo_path, f"http://foo/bar.git", usersync=False)
o.uid == os_data.uid
o.gid == os_data.gid
- @mock.patch('snakeoil.process.spawn.spawn')
+ @mock.patch("snakeoil.process.spawn.spawn")
def test_usersync_portage_perms(self, spawn):
# sync uses portage perms if repo dir doesn't exist
o = base.Syncer(self.repo_path, f"http://foo/bar.git", usersync=True)
o.uid == os_data.portage_uid
o.gid == os_data.portage_gid
- @mock.patch('snakeoil.process.spawn.spawn')
+ @mock.patch("snakeoil.process.spawn.spawn")
def test_usersync_repo_dir_perms(self, spawn):
# and repo dir perms if it does exist
- with mock.patch('os.stat') as stat:
+ with mock.patch("os.stat") as stat:
stat.return_value = mock.Mock(st_uid=1234, st_gid=5678)
o = base.Syncer(self.repo_path, f"http://foo/bar.git", usersync=True)
stat.assert_called()
@@ -61,62 +60,60 @@ class TestSyncer:
assert o.gid == 5678
-@mock.patch('snakeoil.process.find_binary')
+@mock.patch("snakeoil.process.find_binary")
class TestExternalSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = str(tmp_path / 'repo')
+ self.repo_path = str(tmp_path / "repo")
def test_missing_binary(self, find_binary):
- find_binary.side_effect = CommandNotFound('foo')
+ find_binary.side_effect = CommandNotFound("foo")
with pytest.raises(base.MissingBinary):
- base.ExternalSyncer(self.repo_path, 'http://dar')
+ base.ExternalSyncer(self.repo_path, "http://dar")
def test_existing_binary(self, find_binary):
# fake external syncer
class FooSyncer(base.ExternalSyncer):
- binary = 'foo'
+ binary = "foo"
# fake that the external binary exists
find_binary.side_effect = lambda x: x
- o = FooSyncer(self.repo_path, 'http://dar')
- assert o.uri == 'http://dar'
- assert o.binary == 'foo'
+ o = FooSyncer(self.repo_path, "http://dar")
+ assert o.uri == "http://dar"
+ assert o.binary == "foo"
- @mock.patch('snakeoil.process.spawn.spawn')
+ @mock.patch("snakeoil.process.spawn.spawn")
def test_usersync(self, spawn, find_binary):
# fake external syncer
class FooSyncer(base.ExternalSyncer):
- binary = 'foo'
+ binary = "foo"
# fake that the external binary exists
find_binary.side_effect = lambda x: x
- o = FooSyncer(self.repo_path, 'http://dar')
+ o = FooSyncer(self.repo_path, "http://dar")
o.uid = 1234
o.gid = 2345
- o._spawn('cmd', pipes={})
- assert spawn.call_args[1]['uid'] == o.uid
- assert spawn.call_args[1]['gid'] == o.gid
+ o._spawn("cmd", pipes={})
+ assert spawn.call_args[1]["uid"] == o.uid
+ assert spawn.call_args[1]["gid"] == o.gid
-@mock.patch('snakeoil.process.find_binary', return_value='git')
-@mock.patch('snakeoil.process.spawn.spawn')
+@mock.patch("snakeoil.process.find_binary", return_value="git")
+@mock.patch("snakeoil.process.spawn.spawn")
class TestVcsSyncer:
-
def test_basedir_perms_error(self, spawn, find_binary, tmp_path):
- syncer = git.git_syncer(str(tmp_path), 'git://blah.git')
+ syncer = git.git_syncer(str(tmp_path), "git://blah.git")
with pytest.raises(base.PathError):
- with mock.patch('os.stat') as stat:
- stat.side_effect = EnvironmentError('fake exception')
+ with mock.patch("os.stat") as stat:
+ stat.side_effect = EnvironmentError("fake exception")
syncer.sync()
def test_basedir_is_file_error(self, spawn, find_binary, tmp_path):
repo = tmp_path / "repo"
repo.touch()
- syncer = git.git_syncer(str(repo), 'git://blah.git')
+ syncer = git.git_syncer(str(repo), "git://blah.git")
# basedir gets '/' appended by default and stat errors out
with pytest.raises(base.PathError) as excinfo:
@@ -129,32 +126,30 @@ class TestVcsSyncer:
assert "isn't a directory" in str(excinfo.value)
def test_verbose_sync(self, spawn, find_binary, tmp_path):
- syncer = git.git_syncer(str(tmp_path), 'git://blah.git')
+ syncer = git.git_syncer(str(tmp_path), "git://blah.git")
syncer.sync(verbosity=1)
- assert '-v' == spawn.call_args[0][0][-1]
+ assert "-v" == spawn.call_args[0][0][-1]
syncer.sync(verbosity=2)
- assert '-vv' == spawn.call_args[0][0][-1]
+ assert "-vv" == spawn.call_args[0][0][-1]
def test_quiet_sync(self, spawn, find_binary, tmp_path):
- syncer = git.git_syncer(str(tmp_path), 'git://blah.git')
+ syncer = git.git_syncer(str(tmp_path), "git://blah.git")
syncer.sync(verbosity=-1)
- assert '-q' == spawn.call_args[0][0][-1]
+ assert "-q" == spawn.call_args[0][0][-1]
class TestGenericSyncer:
-
def test_init(self):
with pytest.raises(base.UriError):
- base.GenericSyncer('/', 'seriouslynotaprotocol://blah/')
+ base.GenericSyncer("/", "seriouslynotaprotocol://blah/")
- syncer = base.GenericSyncer('/', f'tar+https://blah.tar.gz')
+ syncer = base.GenericSyncer("/", f"tar+https://blah.tar.gz")
assert tar.tar_syncer is syncer.__class__
class TestDisabledSyncer:
-
def test_init(self):
- syncer = base.DisabledSyncer('/foo/bar', f'https://blah.git')
+ syncer = base.DisabledSyncer("/foo/bar", f"https://blah.git")
assert syncer.disabled
# syncing should also be disabled
assert not syncer.uri
@@ -162,14 +157,13 @@ class TestDisabledSyncer:
class TestAutodetectSyncer:
-
def test_no_syncer_detected(self, tmp_path):
syncer = base.AutodetectSyncer(str(tmp_path))
assert isinstance(syncer, base.DisabledSyncer)
- @mock.patch('snakeoil.process.find_binary', return_value='git')
+ @mock.patch("snakeoil.process.find_binary", return_value="git")
def test_syncer_detected(self, find_binary, tmp_path):
- d = tmp_path / '.git'
+ d = tmp_path / ".git"
d.mkdir()
syncer = base.AutodetectSyncer(str(tmp_path))
assert isinstance(syncer, git.git_syncer)
diff --git a/tests/sync/test_bzr.py b/tests/sync/test_bzr.py
index e1b45d4f..f036c647 100644
--- a/tests/sync/test_bzr.py
+++ b/tests/sync/test_bzr.py
@@ -6,10 +6,9 @@ from snakeoil.process import CommandNotFound
class TestBzrSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = tmp_path / 'repo'
+ self.repo_path = tmp_path / "repo"
def test_uri_parse(self):
assert bzr.bzr_syncer.parse_uri("bzr+http://dar") == "http://dar"
@@ -18,13 +17,13 @@ class TestBzrSyncer:
bzr.bzr_syncer.parse_uri("bzr://dar")
# external binary doesn't exist
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.side_effect = CommandNotFound('bzr')
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.side_effect = CommandNotFound("bzr")
with pytest.raises(base.SyncError):
bzr.bzr_syncer(str(self.repo_path), "bzr+http://foon.com/dar")
# fake that the external binary exists
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.return_value = 'bzr'
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.return_value = "bzr"
o = bzr.bzr_syncer(str(self.repo_path), "bzr+http://dar")
o.uri == "http://dar"
diff --git a/tests/sync/test_cvs.py b/tests/sync/test_cvs.py
index d92e5765..7c02619f 100644
--- a/tests/sync/test_cvs.py
+++ b/tests/sync/test_cvs.py
@@ -6,26 +6,26 @@ from snakeoil.process import CommandNotFound
class TestCVSSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = tmp_path / 'repo'
+ self.repo_path = tmp_path / "repo"
def test_uri_parse(self):
# external binary doesn't exist
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.side_effect = CommandNotFound('cvs')
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.side_effect = CommandNotFound("cvs")
with pytest.raises(base.SyncError):
- cvs.cvs_syncer(
- str(self.repo_path), "cvs+/bin/sh://foon.com/dar")
+ cvs.cvs_syncer(str(self.repo_path), "cvs+/bin/sh://foon.com/dar")
# fake that the external binary exists
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.return_value = 'cvs'
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.return_value = "cvs"
# nonexistent rsh
- with mock.patch('pkgcore.sync.base.ExternalSyncer.require_binary') as require_binary:
- require_binary.side_effect = base.MissingBinary('', 'rsh')
+ with mock.patch(
+ "pkgcore.sync.base.ExternalSyncer.require_binary"
+ ) as require_binary:
+ require_binary.side_effect = base.MissingBinary("", "rsh")
with pytest.raises(base.SyncError):
cvs.cvs_syncer(str(self.repo_path), "cvs+rsh://foon.com/dar")
@@ -41,8 +41,10 @@ class TestCVSSyncer:
assert o.rsh == None
assert o.env["CVSROOT"] == ":pserver:dar"
- with mock.patch('pkgcore.sync.base.ExternalSyncer.require_binary') as require_binary:
- require_binary.return_value = '/bin/sh'
+ with mock.patch(
+ "pkgcore.sync.base.ExternalSyncer.require_binary"
+ ) as require_binary:
+ require_binary.return_value = "/bin/sh"
o = cvs.cvs_syncer(str(self.repo_path), "cvs+/bin/sh://dar:module")
assert o.rsh == "/bin/sh"
assert o.uri == ":ext:dar"
diff --git a/tests/sync/test_darcs.py b/tests/sync/test_darcs.py
index 96a7bf71..4abdaa10 100644
--- a/tests/sync/test_darcs.py
+++ b/tests/sync/test_darcs.py
@@ -6,10 +6,9 @@ from snakeoil.process import CommandNotFound
class TestDarcsSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = tmp_path / 'repo'
+ self.repo_path = tmp_path / "repo"
def test_uri_parse(self):
assert darcs.darcs_syncer.parse_uri("darcs+http://dar") == "http://dar"
@@ -18,13 +17,13 @@ class TestDarcsSyncer:
darcs.darcs_syncer.parse_uri("darcs://dar")
# external binary doesn't exist
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.side_effect = CommandNotFound('darcs')
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.side_effect = CommandNotFound("darcs")
with pytest.raises(base.SyncError):
darcs.darcs_syncer(str(self.repo_path), "darcs+http://foon.com/dar")
# fake that the external binary exists
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.return_value = 'bzr'
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.return_value = "bzr"
o = darcs.darcs_syncer(str(self.repo_path), "darcs+http://dar")
assert o.uri == "http://dar"
diff --git a/tests/sync/test_git.py b/tests/sync/test_git.py
index 8ef8c845..16a12b6e 100644
--- a/tests/sync/test_git.py
+++ b/tests/sync/test_git.py
@@ -7,10 +7,9 @@ from snakeoil.process import CommandNotFound
class TestGitSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = tmp_path / 'repo'
+ self.repo_path = tmp_path / "repo"
def test_uri_parse(self):
assert git.git_syncer.parse_uri("git+http://dar") == "http://dar"
@@ -19,42 +18,42 @@ class TestGitSyncer:
git.git_syncer.parse_uri("git+://dar")
# external binary doesn't exist
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.side_effect = CommandNotFound('git')
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.side_effect = CommandNotFound("git")
with pytest.raises(base.SyncError):
git.git_syncer(str(self.repo_path), "git+http://foon.com/dar")
# fake that the external binary exists
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.return_value = 'git'
- for proto in ('http', 'https'):
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.return_value = "git"
+ for proto in ("http", "https"):
for uri in (f"git+{proto}://repo.git", f"{proto}://repo.git"):
o = git.git_syncer(str(self.repo_path), uri)
assert o.uri == f"{proto}://repo.git"
- @mock.patch('snakeoil.process.spawn.spawn')
+ @mock.patch("snakeoil.process.spawn.spawn")
def test_sync(self, spawn):
- uri = 'git://foo.git'
- with mock.patch('snakeoil.process.find_binary', return_value='git'):
+ uri = "git://foo.git"
+ with mock.patch("snakeoil.process.find_binary", return_value="git"):
syncer = git.git_syncer(str(self.repo_path), uri)
# initial sync
syncer.sync()
assert spawn.call_args[0] == (
- ['git', 'clone', uri, str(self.repo_path) + os.path.sep],)
- assert spawn.call_args[1]['cwd'] is None
+ ["git", "clone", uri, str(self.repo_path) + os.path.sep],
+ )
+ assert spawn.call_args[1]["cwd"] is None
# repo update
self.repo_path.mkdir()
syncer.sync()
- assert spawn.call_args[0] == (['git', 'pull'],)
- assert spawn.call_args[1]['cwd'] == syncer.basedir
+ assert spawn.call_args[0] == (["git", "pull"],)
+ assert spawn.call_args[1]["cwd"] == syncer.basedir
@pytest.mark_network
class TestGitSyncerReal:
-
def test_sync(self, tmp_path):
- path = tmp_path / 'repo'
+ path = tmp_path / "repo"
syncer = git.git_syncer(str(path), "https://github.com/pkgcore/pkgrepo.git")
assert syncer.sync()
- assert os.path.exists(os.path.join(path, 'metadata', 'layout.conf'))
+ assert os.path.exists(os.path.join(path, "metadata", "layout.conf"))
assert syncer.sync()
diff --git a/tests/sync/test_git_svn.py b/tests/sync/test_git_svn.py
index 8dc8cd8e..a79c15cc 100644
--- a/tests/sync/test_git_svn.py
+++ b/tests/sync/test_git_svn.py
@@ -8,10 +8,9 @@ from snakeoil.process import CommandNotFound
class TestGitSVNSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = tmp_path / 'repo'
+ self.repo_path = tmp_path / "repo"
def test_uri_parse(self):
assert git_svn.git_svn_syncer.parse_uri("git+svn+http://dar") == "http://dar"
@@ -20,13 +19,15 @@ class TestGitSVNSyncer:
git_svn.git_svn_syncer.parse_uri("git+svn+://dar")
# external binary doesn't exist
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.side_effect = CommandNotFound('git')
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.side_effect = CommandNotFound("git")
with pytest.raises(base.SyncError):
- git_svn.git_svn_syncer(str(self.repo_path), "git+svn+http://foon.com/dar")
+ git_svn.git_svn_syncer(
+ str(self.repo_path), "git+svn+http://foon.com/dar"
+ )
# fake that the external binary exists
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.return_value = 'git'
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.return_value = "git"
o = git_svn.git_svn_syncer(str(self.repo_path), "git+svn+http://dar")
assert o.uri == "http://dar"
diff --git a/tests/sync/test_hg.py b/tests/sync/test_hg.py
index 59904af9..490eac05 100644
--- a/tests/sync/test_hg.py
+++ b/tests/sync/test_hg.py
@@ -7,10 +7,9 @@ from snakeoil.process import CommandNotFound
class TestHgSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = tmp_path / 'repo'
+ self.repo_path = tmp_path / "repo"
def test_uri_parse(self):
assert hg.hg_syncer.parse_uri("hg+http://dar") == "http://dar"
@@ -20,30 +19,31 @@ class TestHgSyncer:
hg.hg_syncer.parse_uri("hg://dar")
# external binary doesn't exist
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.side_effect = CommandNotFound('svn')
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.side_effect = CommandNotFound("svn")
with pytest.raises(base.SyncError):
hg.hg_syncer(str(self.repo_path), "hg+http://foon.com/dar")
# fake that the external binary exists
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.return_value = 'hg'
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.return_value = "hg"
o = hg.hg_syncer(str(self.repo_path), "hg+http://dar")
assert o.uri == "http://dar"
- @mock.patch('snakeoil.process.spawn.spawn')
+ @mock.patch("snakeoil.process.spawn.spawn")
def test_sync(self, spawn):
- uri = 'https://foo/bar'
- with mock.patch('snakeoil.process.find_binary', return_value='hg'):
- syncer = hg.hg_syncer(str(self.repo_path), f'hg+{uri}')
+ uri = "https://foo/bar"
+ with mock.patch("snakeoil.process.find_binary", return_value="hg"):
+ syncer = hg.hg_syncer(str(self.repo_path), f"hg+{uri}")
# initial sync
syncer.sync()
assert spawn.call_args[0] == (
- ['hg', 'clone', uri, str(self.repo_path) + os.path.sep],)
- assert spawn.call_args[1]['cwd'] is None
+ ["hg", "clone", uri, str(self.repo_path) + os.path.sep],
+ )
+ assert spawn.call_args[1]["cwd"] is None
# repo update
self.repo_path.mkdir()
syncer.sync()
- assert spawn.call_args[0] == (['hg', 'pull', '-u', uri],)
- assert spawn.call_args[1]['cwd'] == syncer.basedir
+ assert spawn.call_args[0] == (["hg", "pull", "-u", uri],)
+ assert spawn.call_args[1]["cwd"] == syncer.basedir
diff --git a/tests/sync/test_rsync.py b/tests/sync/test_rsync.py
index 14bfc59f..8bfba18f 100644
--- a/tests/sync/test_rsync.py
+++ b/tests/sync/test_rsync.py
@@ -10,41 +10,39 @@ from snakeoil.process import CommandNotFound
def fake_ips(num):
"""Generate simple IPv4 addresses given the amount to create."""
- return [
- (None, None, None, None, ('.'.join(str(x) * 4), 0))
- for x in range(num)
- ]
+ return [(None, None, None, None, (".".join(str(x) * 4), 0)) for x in range(num)]
-@mock.patch('socket.getaddrinfo', return_value=fake_ips(3))
-@mock.patch('snakeoil.process.spawn.spawn')
+@mock.patch("socket.getaddrinfo", return_value=fake_ips(3))
+@mock.patch("snakeoil.process.spawn.spawn")
class TestRsyncSyncer:
_syncer_class = rsync.rsync_syncer
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = str(tmp_path / 'repo')
- with mock.patch('snakeoil.process.find_binary', return_value='rsync'):
+ self.repo_path = str(tmp_path / "repo")
+ with mock.patch("snakeoil.process.find_binary", return_value="rsync"):
self.syncer = self._syncer_class(
- self.repo_path, "rsync://rsync.gentoo.org/gentoo-portage")
+ self.repo_path, "rsync://rsync.gentoo.org/gentoo-portage"
+ )
- @mock.patch('snakeoil.process.find_binary')
+ @mock.patch("snakeoil.process.find_binary")
def test_uri_parse_rsync_missing(self, find_binary, spawn, getaddrinfo):
- find_binary.side_effect = CommandNotFound('rsync')
+ find_binary.side_effect = CommandNotFound("rsync")
with pytest.raises(base.SyncError):
- self._syncer_class(self.repo_path, 'rsync://foon.com/dar')
+ self._syncer_class(self.repo_path, "rsync://foon.com/dar")
- @mock.patch('snakeoil.process.find_binary')
+ @mock.patch("snakeoil.process.find_binary")
def test_uri_parse(self, find_binary, spawn, getaddrinfo):
find_binary.side_effect = lambda x: x
- o = self._syncer_class(self.repo_path, 'rsync://dar/module')
- assert o.uri == 'rsync://dar/module/'
+ o = self._syncer_class(self.repo_path, "rsync://dar/module")
+ assert o.uri == "rsync://dar/module/"
assert o.rsh == None
- o = self._syncer_class(self.repo_path, 'rsync+/bin/sh://dar/module')
- assert o.uri == 'rsync://dar/module/'
- assert o.rsh == '/bin/sh'
+ o = self._syncer_class(self.repo_path, "rsync+/bin/sh://dar/module")
+ assert o.uri == "rsync://dar/module/"
+ assert o.rsh == "/bin/sh"
def test_successful_sync(self, spawn, getaddrinfo):
spawn.return_value = 0
@@ -55,21 +53,21 @@ class TestRsyncSyncer:
spawn.return_value = 1
with pytest.raises(base.SyncError) as excinfo:
assert self.syncer.sync()
- assert str(excinfo.value).startswith('rsync command syntax error:')
+ assert str(excinfo.value).startswith("rsync command syntax error:")
spawn.assert_called_once()
def test_failed_disk_space_sync(self, spawn, getaddrinfo):
spawn.return_value = 11
with pytest.raises(base.SyncError) as excinfo:
assert self.syncer.sync()
- assert str(excinfo.value) == 'rsync ran out of disk space'
+ assert str(excinfo.value) == "rsync ran out of disk space"
spawn.assert_called_once()
def test_retried_sync(self, spawn, getaddrinfo):
spawn.return_value = 99
with pytest.raises(base.SyncError) as excinfo:
assert self.syncer.sync()
- assert str(excinfo.value) == 'all attempts failed'
+ assert str(excinfo.value) == "all attempts failed"
# rsync should retry every resolved IP related to the sync URI
assert len(spawn.mock_calls) == 3
@@ -79,14 +77,14 @@ class TestRsyncSyncer:
getaddrinfo.return_value = fake_ips(self.syncer.retries + 1)
with pytest.raises(base.SyncError) as excinfo:
assert self.syncer.sync()
- assert str(excinfo.value) == 'all attempts failed'
+ assert str(excinfo.value) == "all attempts failed"
assert len(spawn.mock_calls) == self.syncer.retries
def test_failed_dns_sync(self, spawn, getaddrinfo):
getaddrinfo.side_effect = OSError()
with pytest.raises(base.SyncError) as excinfo:
assert self.syncer.sync()
- assert str(excinfo.value).startswith('DNS resolution failed')
+ assert str(excinfo.value).startswith("DNS resolution failed")
spawn.assert_not_called()
@@ -97,21 +95,23 @@ class TestRsyncTimestampSyncer(TestRsyncSyncer):
@pytest.mark_network
class TestRsyncSyncerReal:
-
def test_sync(self, tmp_path):
# perform a tarball sync for initial week-old base
- path = tmp_path / 'repo'
+ path = tmp_path / "repo"
week_old = datetime.datetime.now() - datetime.timedelta(days=7)
date_str = week_old.strftime("%Y%m%d")
syncer = tar_syncer(
- str(path), f"http://distfiles.gentoo.org/snapshots/portage-{date_str}.tar.xz")
+ str(path),
+ f"http://distfiles.gentoo.org/snapshots/portage-{date_str}.tar.xz",
+ )
assert syncer.sync()
- timestamp = os.path.join(path, 'metadata', 'timestamp.chk')
+ timestamp = os.path.join(path, "metadata", "timestamp.chk")
assert os.path.exists(timestamp)
stat = os.stat(timestamp)
# run rsync over the unpacked repo tarball to update to the latest tree
syncer = rsync.rsync_timestamp_syncer(
- str(path), "rsync://rsync.gentoo.org/gentoo-portage")
+ str(path), "rsync://rsync.gentoo.org/gentoo-portage"
+ )
assert syncer.sync()
assert stat != os.stat(timestamp)
diff --git a/tests/sync/test_sqfs.py b/tests/sync/test_sqfs.py
index 47b53058..88e23da2 100644
--- a/tests/sync/test_sqfs.py
+++ b/tests/sync/test_sqfs.py
@@ -6,9 +6,10 @@ from pkgcore.sync.sqfs import sqfs_syncer
class TestSqfsSyncer:
-
def test_uri_parse(self):
- assert sqfs_syncer.parse_uri("sqfs+http://repo.lzo.sqfs") == "http://repo.lzo.sqfs"
+ assert (
+ sqfs_syncer.parse_uri("sqfs+http://repo.lzo.sqfs") == "http://repo.lzo.sqfs"
+ )
# missing actual URI protocol
with pytest.raises(base.UriError):
@@ -24,12 +25,12 @@ class TestSqfsSyncer:
@pytest.mark_network
class TestSqfsSyncerReal:
-
def test_sync(self, tmp_path):
- path = tmp_path / 'repo'
+ path = tmp_path / "repo"
syncer = sqfs_syncer(
- str(path),
- "sqfs+http://distfiles.gentoo.org/snapshots/squashfs/gentoo-current.lzo.sqfs")
+ str(path),
+ "sqfs+http://distfiles.gentoo.org/snapshots/squashfs/gentoo-current.lzo.sqfs",
+ )
assert syncer.sync()
sqfs = os.path.join(syncer.basedir, syncer.basename)
assert os.path.exists(sqfs)
diff --git a/tests/sync/test_svn.py b/tests/sync/test_svn.py
index 77757b22..1edeffc2 100644
--- a/tests/sync/test_svn.py
+++ b/tests/sync/test_svn.py
@@ -6,23 +6,22 @@ from snakeoil.process import CommandNotFound
class TestSVNSyncer:
-
@pytest.fixture(autouse=True)
def _setup(self, tmp_path):
- self.repo_path = tmp_path / 'repo'
+ self.repo_path = tmp_path / "repo"
def test_uri_parse(self):
with pytest.raises(base.UriError):
svn.svn_syncer.parse_uri("svn+://dar")
# external binary doesn't exist
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.side_effect = CommandNotFound('svn')
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.side_effect = CommandNotFound("svn")
with pytest.raises(base.SyncError):
svn.svn_syncer(str(self.repo_path), "svn+http://foon.com/dar")
# fake that the external binary exists
- with mock.patch('snakeoil.process.find_binary') as find_binary:
- find_binary.return_value = 'svn'
+ with mock.patch("snakeoil.process.find_binary") as find_binary:
+ find_binary.return_value = "svn"
o = svn.svn_syncer(str(self.repo_path), "svn+http://dar")
assert o.uri == "http://dar"
diff --git a/tests/sync/test_tar.py b/tests/sync/test_tar.py
index e8be1995..0e65ea56 100644
--- a/tests/sync/test_tar.py
+++ b/tests/sync/test_tar.py
@@ -6,7 +6,6 @@ from pkgcore.sync.tar import tar_syncer
class TestTarSyncer:
-
def test_uri_parse(self):
assert tar_syncer.parse_uri("tar+http://repo.tar.gz") == "http://repo.tar.gz"
@@ -19,7 +18,7 @@ class TestTarSyncer:
tar_syncer.parse_uri("tar+https://repo.tar.foo")
for ext in tar_syncer.supported_exts:
- for proto in ('http', 'https'):
+ for proto in ("http", "https"):
for uri in (f"tar+{proto}://repo{ext}", f"{proto}://repo{ext}"):
o = tar_syncer("/tmp/foon", uri)
assert o.uri == f"{proto}://repo{ext}"
@@ -27,13 +26,13 @@ class TestTarSyncer:
@pytest.mark_network
class TestTarSyncerReal:
-
def test_sync(self, tmp_path):
- path = tmp_path / 'repo'
+ path = tmp_path / "repo"
syncer = tar_syncer(
- str(path), "https://github.com/pkgcore/pkgrepo/archive/master.tar.gz")
+ str(path), "https://github.com/pkgcore/pkgrepo/archive/master.tar.gz"
+ )
assert syncer.sync()
- layout_conf = os.path.join(path, 'metadata', 'layout.conf')
+ layout_conf = os.path.join(path, "metadata", "layout.conf")
assert os.path.exists(layout_conf)
stat = os.stat(layout_conf)
# re-sync and verify that the repo didn't get replaced