diff options
author | 2022-12-25 11:26:31 -0800 | |
---|---|---|
committer | 2022-12-25 21:32:26 +0200 | |
commit | b17af120f9081aa30d6a368fd75a69c298cb70e0 (patch) | |
tree | 3de8bc1fba4fbd004b9b1703e67b1a9feb5ab6fc /tests/sync | |
parent | Fix secondary exception from when an EBD fails to start. (diff) | |
download | pkgcore-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.py | 76 | ||||
-rw-r--r-- | tests/sync/test_bzr.py | 11 | ||||
-rw-r--r-- | tests/sync/test_cvs.py | 26 | ||||
-rw-r--r-- | tests/sync/test_darcs.py | 11 | ||||
-rw-r--r-- | tests/sync/test_git.py | 33 | ||||
-rw-r--r-- | tests/sync/test_git_svn.py | 15 | ||||
-rw-r--r-- | tests/sync/test_hg.py | 28 | ||||
-rw-r--r-- | tests/sync/test_rsync.py | 56 | ||||
-rw-r--r-- | tests/sync/test_sqfs.py | 13 | ||||
-rw-r--r-- | tests/sync/test_svn.py | 11 | ||||
-rw-r--r-- | tests/sync/test_tar.py | 11 |
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 |