From 44fbaf36e86c14481fcd4e0b33c207910a96e03e Mon Sep 17 00:00:00 2001 From: Jean-Louis Fuchs Date: Tue, 16 Jan 2024 13:34:53 +0100 Subject: [PATCH 1/3] test: move last test from aptly_test.py --- pyaptly/__init__.py | 6 +++++- pyaptly/aptly_test.py | 2 ++ pyaptly/tests/test_publish.py | 19 +++++++++++++++++++ 3 files changed, 26 insertions(+), 1 deletion(-) diff --git a/pyaptly/__init__.py b/pyaptly/__init__.py index 8d96749..27b5644 100644 --- a/pyaptly/__init__.py +++ b/pyaptly/__init__.py @@ -3,4 +3,8 @@ Configuration is based on toml input files. """ -from pyaptly.legacy import SystemStateReader, main # type: ignore # TODO # noqa: F401 +from pyaptly.legacy import ( # type: ignore # TODO # noqa: F401 + Command, + SystemStateReader, + main, +) diff --git a/pyaptly/aptly_test.py b/pyaptly/aptly_test.py index 492335c..2f8a6e2 100644 --- a/pyaptly/aptly_test.py +++ b/pyaptly/aptly_test.py @@ -39,6 +39,7 @@ def mock_subprocess(): output.stop() +@pytest.mark.skip # mock_subprocess was never used def test_debug(): """Test if debug is enabled with -d""" with mock_subprocess() as (_, gpg): @@ -54,6 +55,7 @@ def test_debug(): assert logging.getLogger().level == logging.DEBUG +@pytest.mark.skip def test_pretend(): """Test if pretend is enabled with -p""" with test.clean_and_config( diff --git a/pyaptly/tests/test_publish.py b/pyaptly/tests/test_publish.py index 81d9f4a..1c08228 100644 --- a/pyaptly/tests/test_publish.py +++ b/pyaptly/tests/test_publish.py @@ -39,6 +39,25 @@ def test_repo_crate_basic(config, repo_create): pass +@pytest.mark.parametrize("config", ["publish.toml"], indirect=True) +def test_pretend(config, snapshot_create, test_key_03): + """Test if pretend is enabled with -p.""" + args = [ + "-p", + "-c", + config, + "publish", + "create", + "fakerepo01", + ] + pyaptly.main(args) + state = pyaptly.SystemStateReader() + state.read() + assert set() == state.publishes + assert {} == state.publish_map + assert pyaptly.Command.pretend_mode + + @pytest.mark.parametrize("config", ["publish-repo.toml"], indirect=True) def test_publish_create_repo(config, repo_create): """Test if creating repo publishes works.""" From a464151cc955e6c9b340c6bc7b18522e4f5da52f Mon Sep 17 00:00:00 2001 From: Jean-Louis Fuchs Date: Tue, 16 Jan 2024 13:35:27 +0100 Subject: [PATCH 2/3] test: remove aptly_test.py --- pyaptly/aptly_test.py | 922 ------------------------------------------ 1 file changed, 922 deletions(-) delete mode 100644 pyaptly/aptly_test.py diff --git a/pyaptly/aptly_test.py b/pyaptly/aptly_test.py deleted file mode 100644 index 2f8a6e2..0000000 --- a/pyaptly/aptly_test.py +++ /dev/null @@ -1,922 +0,0 @@ -# type: ignore # TODO -# flake8: noqa # TODO - -"""Testing pyaptly""" -import contextlib -import logging -import os - -import freezegun -import pytest -import testfixtures - -from pyaptly.legacy import ( - Command, - SystemStateReader, - call_output, - main, - snapshot_spec_to_name, -) - -from . import test - -try: - import unittest.mock as mock -except ImportError: # pragma: no cover - import mock - - -_test_base = os.path.dirname(os.path.abspath(__file__)).encode("UTF-8") - - -@contextlib.contextmanager -def mock_subprocess(): - """Mock subprocess that no commands are executed""" - call = mock.patch("subprocess.check_call") - output = mock.patch("pyaptly.legacy.call_output") - yield (call.start(), output.start()) - call.stop() - output.stop() - - -@pytest.mark.skip # mock_subprocess was never used -def test_debug(): - """Test if debug is enabled with -d""" - with mock_subprocess() as (_, gpg): - gpg.side_effect = lambda _: ("", "") - args = [ - "-d", - "-c", - os.path.join(_test_base, b"test01.yml").decode("UTF-8"), - "mirror", - "create", - ] - main(args) - assert logging.getLogger().level == logging.DEBUG - - -@pytest.mark.skip -def test_pretend(): - """Test if pretend is enabled with -p""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish.yml", - ) - ) as (tyml, config): - do_snapshot_create(config) - args = [ - "-p", - "-c", - config, - "publish", - "create", - "fakerepo01", - ] - main(args) - state = SystemStateReader() - state.read() - assert set() == state.publishes - assert {} == state.publish_map - assert Command.pretend_mode - - -@pytest.mark.skip -def test_mirror_create(): - """Test if createing mirrors works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"mirror-google.yml", - ) - ) as (tyml, config): - args = ["-c", config, "mirror", "create"] - keys_added = [] - with testfixtures.LogCapture() as l: - main(args) - for rec in l.records: - for arg in rec.args: - if isinstance(arg, list): - if arg[0] == "gpg": - keys_added.append(arg[7]) - assert len(keys_added) > 0 - assert len(keys_added) == len(set(keys_added)), "Key multiple times added" - - expect = set(tyml["mirror"].keys()) - state = SystemStateReader() - state.read() - assert state.mirrors == expect - - -def do_mirror_update(config): # pragma: no cover - """Test if updating mirrors works.""" - args = ["-c", config, "mirror", "create"] - state = SystemStateReader() - state.read() - assert "fakerepo01" not in state.mirrors - main(args) - state.read() - assert "fakerepo01" in state.mirrors - args[3] = "update" - main(args) - args = [ - "aptly", - "mirror", - "show", - ] - args01 = list(args) - args01.append("fakerepo01") - aptly_state = test.execute_and_parse_show_cmd(args01) - assert aptly_state["number of packages"] == "2" - - -@pytest.mark.skip -def test_mirror_update(): - """Test if updating mirrors works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"mirror-no-google.yml", - ) - ) as (tyml, config): - do_mirror_update(config) - - -@pytest.mark.skip -def test_mirror_update_inexistent(): - """Test if updating an inexistent mirror causes an error.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"mirror-no-google.yml", - ) - ) as (tyml, config): - do_mirror_update(config) - args = ["-c", config, "mirror", "update", "asdfasdf"] - error = False - try: - main(args) - except ValueError: - error = True - assert error - - -@pytest.mark.skip -def test_mirror_update_single(): - """Test if updating a single mirror works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"mirror-no-google.yml", - ) - ) as (tyml, config): - do_mirror_update(config) - args = ["-c", config, "mirror", "update", "fakerepo01"] - main(args) - - -def do_snapshot_create(config): # pragma: no cover - """Test if createing snapshots works""" - do_mirror_update(config) - args = ["-c", config, "snapshot", "create"] - main(args) - state = SystemStateReader() - state.read() - assert set(["fakerepo01-20121010T0000Z", "fakerepo02-20121006T0000Z"]).issubset( - state.snapshots - ) - return state - - -@pytest.mark.skip -def test_snapshot_create_inexistent(): - """Test if creating an inexistent snapshot raises an error.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot.yml", - ) - ) as (tyml, config): - do_mirror_update(config) - args = [ - "-c", - config, - "snapshot", - "create", - "asdfasdf-%T", - ] - error = False - try: - main(args) - except ValueError: - error = True - assert error - - -@pytest.mark.skip -def test_snapshot_create_single(): - """Test if single snapshot create works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot.yml", - ) - ) as (tyml, config): - do_mirror_update(config) - args = [ - "-c", - config, - "snapshot", - "create", - "fakerepo01-%T", - ] - main(args) - state = SystemStateReader() - state.read() - assert set(["fakerepo01-20121010T0000Z"]).issubset(state.snapshots) - - -@pytest.mark.skip -def test_snapshot_create_rotating(): - """Test if rotating snapshot create works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot-current.yml", - ) - ) as (tyml, config): - do_mirror_update(config) - args = [ - "-c", - config, - "snapshot", - "create", - ] - main(args) - state = SystemStateReader() - state.read() - assert set( - [ - "fake-current", - "fakerepo01-current", - "fakerepo02-current", - ] - ).issubset(state.snapshots) - - -@pytest.mark.skip -def test_snapshot_update_rotating(): - """Test if rotating snapshot update works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot-current.yml", - ) - ) as (tyml, config): - do_snapshot_update_rotating(config) - - -@pytest.mark.skip -def test_snapshot_update_threetimes_rotating(): - """Test if rotating snapshot update works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot-current.yml", - ) - ) as (tyml, config): - do_snapshot_update_rotating(config) - with freezegun.freeze_time("2012-10-11 10:10:10"): - args = [ - "-c", - config, - "snapshot", - "update", - ] - main(args) - state = SystemStateReader() - state.read() - assert set( - [ - "fake-current", - "fakerepo01-current-rotated-20121010T1010Z", - "fakerepo02-current-rotated-20121010T1010Z", - "fakerepo01-current-rotated-20121011T1010Z", - "fakerepo02-current-rotated-20121011T1010Z", - ] - ).issubset(state.snapshots) - expected = { - "fake-current": set(["fakerepo01-current", "fakerepo02-current"]), - "fake-current-rotated-20121010T1010Z": set( - [ - "fakerepo01-current-rotated-20121010T1010Z", - "fakerepo02-current-rotated-20121010T1010Z", - ] - ), - "fake-current-rotated-20121011T1010Z": set( - [ - "fakerepo01-current-rotated-20121011T1010Z", - "fakerepo02-current-rotated-20121011T1010Z", - ] - ), - "fakerepo01-current": set([]), - "fakerepo01-current-rotated-20121010T1010Z": set([]), - "fakerepo01-current-rotated-20121011T1010Z": set([]), - "fakerepo02-current": set([]), - "fakerepo02-current-rotated-20121010T1010Z": set([]), - "fakerepo02-current-rotated-20121011T1010Z": set([]), - } - assert state.snapshot_map == expected - - with freezegun.freeze_time("2012-10-12 10:10:10"): - args = [ - "-c", - config, - "snapshot", - "update", - ] - main(args) - state = SystemStateReader() - state.read() - assert set( - [ - "fake-current", - "fakerepo01-current-rotated-20121011T1010Z", - "fakerepo02-current-rotated-20121011T1010Z", - "fakerepo01-current-rotated-20121012T1010Z", - "fakerepo02-current-rotated-20121012T1010Z", - ] - ).issubset(state.snapshots) - expected = { - "fake-current": set(["fakerepo01-current", "fakerepo02-current"]), - "fake-current-rotated-20121010T1010Z": set( - [ - "fakerepo01-current-rotated-20121010T1010Z", - "fakerepo02-current-rotated-20121010T1010Z", - ] - ), - "fake-current-rotated-20121011T1010Z": set( - [ - "fakerepo01-current-rotated-20121011T1010Z", - "fakerepo02-current-rotated-20121011T1010Z", - ] - ), - "fake-current-rotated-20121012T1010Z": set( - [ - "fakerepo01-current-rotated-20121012T1010Z", - "fakerepo02-current-rotated-20121012T1010Z", - ] - ), - "fakerepo01-current": set([]), - "fakerepo01-current-rotated-20121010T1010Z": set([]), - "fakerepo01-current-rotated-20121011T1010Z": set([]), - "fakerepo01-current-rotated-20121012T1010Z": set([]), - "fakerepo02-current": set([]), - "fakerepo02-current-rotated-20121010T1010Z": set([]), - "fakerepo02-current-rotated-20121011T1010Z": set([]), - "fakerepo02-current-rotated-20121012T1010Z": set([]), - } - assert state.snapshot_map == expected - - -def do_snapshot_update_rotating(config): # pragma: no cover - """Helper for rotating snapshot tests""" - do_mirror_update(config) - args = [ - "-c", - config, - "snapshot", - "create", - ] - main(args) - state = SystemStateReader() - state.read() - assert set( - [ - "fake-current", - "fakerepo01-current", - "fakerepo02-current", - ] - ).issubset(state.snapshots) - args = [ - "-c", - config, - "snapshot", - "update", - ] - main(args) - state.read() - assert set( - [ - "fake-current", - "fakerepo01-current-rotated-20121010T1010Z", - "fakerepo02-current-rotated-20121010T1010Z", - ] - ).issubset(state.snapshots) - expected = { - "fake-current": set(["fakerepo01-current", "fakerepo02-current"]), - "fake-current-rotated-20121010T1010Z": set( - [ - "fakerepo01-current-rotated-20121010T1010Z", - "fakerepo02-current-rotated-20121010T1010Z", - ] - ), - "fakerepo01-current": set([]), - "fakerepo01-current-rotated-20121010T1010Z": set([]), - "fakerepo02-current": set([]), - "fakerepo02-current-rotated-20121010T1010Z": set([]), - } - assert state.snapshot_map == expected - - -@pytest.mark.skip -def test_snapshot_create_basic(): - """Test if snapshot create works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot.yml", - ) - ) as (tyml, config): - state = do_snapshot_create(config) - assert ( - set(["fakerepo01-20121010T0000Z", "fakerepo02-20121006T0000Z"]) - == state.snapshots - ) - - -@pytest.mark.skip -def test_snapshot_create_repo(): - """Test if repo snapshot create works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot_repo.yml", - ) - ) as (tyml, config): - do_repo_create(config) - args = ["-c", config, "snapshot", "create"] - main(args) - state = SystemStateReader() - state.read() - assert set(["centrify-latest"]).issubset(state.snapshots) - return state - - -@pytest.mark.skip -def test_snapshot_create_merge(): - """Test if snapshot merge create works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot_merge.yml", - ) - ) as (tyml, config): - state = do_snapshot_create(config) - assert ( - set( - [ - "fakerepo01-20121010T0000Z", - "fakerepo02-20121006T0000Z", - "superfake-20121010T0000Z", - ] - ) - == state.snapshots - ) - expect = { - "fakerepo01-20121010T0000Z": set([]), - "fakerepo02-20121006T0000Z": set([]), - "superfake-20121010T0000Z": set( - ["fakerepo01-20121010T0000Z", "fakerepo02-20121006T0000Z"] - ), - } - assert expect == state.snapshot_map - - -@pytest.mark.skip -def test_snapshot_create_filter(): - """Test if snapshot filter create works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"snapshot_filter.yml", - ) - ) as (tyml, config): - do_snapshot_create(config) - data, _ = call_output( - ["aptly", "snapshot", "search", "filterfake01-20121010T0000Z", "Name (% *)"] - ) - state = [x.strip() for x in data.split("\n") if x] - expect = ["libhello_0.1-1_amd64"] - assert state == expect - - -def do_publish_create(config): # pragma: no cover - """Test if creating publishes works.""" - do_snapshot_create(config) - args = ["-c", config, "publish", "create"] - main(args) - state = SystemStateReader() - state.read() - assert set(["fakerepo02 main", "fakerepo01 main"]) == state.publishes - expect = { - "fakerepo02 main": set(["fakerepo02-20121006T0000Z"]), - "fakerepo01 main": set(["fakerepo01-20121010T0000Z"]), - } - assert expect == state.publish_map - - -def do_publish_create_rotating(config): # pragma: no cover - """Test if creating publishes works.""" - do_snapshot_update_rotating(config) - args = ["-c", config, "publish", "create"] - main(args) - state = SystemStateReader() - state.read() - assert ( - set( - [ - "fakerepo01/current stable", - "fake/current stable", - "fakerepo02/current stable", - ] - ) - == state.publishes - ) - expect = { - "fake/current stable": set(["fake-current"]), - "fakerepo01/current stable": set(["fakerepo01-current"]), - "fakerepo02/current stable": set(["fakerepo02-current"]), - } - assert expect == state.publish_map - - -@pytest.mark.skip -def test_publish_create_single(): - """Test if creating a single publish works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish.yml", - ), - sign=True, - ) as (tyml, config): - do_snapshot_create(config) - args = [ - "-c", - config, - "publish", - "create", - "fakerepo01", - ] - main(args) - state = SystemStateReader() - state.read() - assert set(["fakerepo01 main"]) == state.publishes - expect = {"fakerepo01 main": set(["fakerepo01-20121010T0000Z"])} - assert expect == state.publish_map - - -@pytest.mark.skip -def test_publish_create_inexistent(): - """Test if creating inexistent publish raises an error.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish.yml", - ) - ) as (tyml, config): - do_snapshot_create(config) - args = [ - "-c", - config, - "publish", - "create", - "asdfasdf", - ] - error = False - try: - main(args) - except ValueError: - error = True - assert error - - -@pytest.mark.skip -def test_publish_create_repo(): - """Test if creating repo publishes works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish_repo.yml", - ), - sign=True, - ) as (tyml, config): - do_repo_create(config) - args = [ - "-c", - config, - "publish", - "create", - ] - main(args) - args = [ - "-c", - config, - "publish", - "update", - ] - main(args) - state = SystemStateReader() - state.read() - assert set(["centrify latest"]) == state.publishes - assert {"centrify latest": set([])} == state.publish_map - - -@pytest.mark.skip -def test_publish_create_basic(): - """Test if creating publishes works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish.yml", - ), - sign=True, - ) as (tyml, config): - do_publish_create(config) - - -@pytest.mark.skip -def test_publish_update_rotating(): - """Test if update rotating publishes works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish-current.yml", - ), - sign=True, - ) as (tyml, config): - do_publish_create_rotating(config) - with freezegun.freeze_time("2012-10-11 10:10:10"): - args = [ - "-c", - config, - "publish", - "update", - ] - main(args) - state = SystemStateReader() - state.read() - expect = { - "fake/current stable": set(["fake-current"]), - "fakerepo01/current stable": set(["fakerepo01-current"]), - "fakerepo02/current stable": set(["fakerepo02-current"]), - } - assert expect == state.publish_map - - -@pytest.mark.skip -def test_publish_snapshot_update_rotating(): - """Test if update rotating publishes via snapshot works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish-current.yml", - ), - sign=True, - ) as (tyml, config): - do_publish_create_rotating(config) - with freezegun.freeze_time("2012-10-11 10:10:10"): - args = [ - "-c", - config, - "snapshot", - "update", - ] - main(args) - state = SystemStateReader() - state.read() - expect = { - "fake/current stable": set(["fake-current"]), - "fakerepo01/current stable": set(["fakerepo01-current"]), - "fakerepo02/current stable": set(["fakerepo02-current"]), - } - assert expect == state.publish_map - - -@pytest.mark.skip -def test_publish_create_rotating(): - """Test if creating rotating publishes works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish-current.yml", - ), - sign=True, - ) as (tyml, config): - do_publish_create_rotating(config) - - -def do_publish_create_republish(config): # pragma: no cover - """Test if creating republishes works.""" - with testfixtures.LogCapture() as l: - do_publish_create(config) - found = False - for rec in l.records: - if rec.levelname == "CRITICAL": - if "has been deferred" in rec.msg: - found = True - assert found - args = [ - "-c", - config, - "publish", - "create", - ] - main(args) - state = SystemStateReader() - state.read() - assert "fakerepo01-stable main" in state.publishes - - -@pytest.mark.skip -def test_publish_create_republish(): - """Test if creating republishes works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish_publish.yml", - ), - sign=True, - ) as (tyml, config): - do_publish_create_republish(config) - - -@pytest.mark.skip -def test_publish_update_republish(): - """Test if update republishes works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish_publish.yml", - ), - sign=True, - ) as (tyml, config): - do_publish_create_republish(config) - with freezegun.freeze_time("2012-10-11 10:10:10"): - args = [ - "-c", - config, - "snapshot", - "create", - ] - main(args) - args = [ - "-c", - config, - "publish", - "update", - ] - main(args) - state = SystemStateReader() - state.read() - assert "fakerepo01-stable main" in state.publishes - # As you see fakerepo01-stable main points to the old snapshot - # this is theoretically not correct, but it will be fixed with - # the next call to publish update. Since we use this from a hourly cron - # job it is no problem. - # This can't be easily fixed and would need a rewrite of the - # dependencies engine. - expect = { - "fakerepo01-stable main": set(["fakerepo01-20121010T0000Z"]), - "fakerepo02 main": set(["fakerepo02-20121006T0000Z"]), - "fakerepo01 main": set(["fakerepo01-20121011T0000Z"]), - } - assert expect == state.publish_map - - -@pytest.mark.skip -def test_publish_updating_basic(): - """Test if updating publishes works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"publish.yml", - ), - sign=True, - ) as (tyml, config): - do_publish_create(config) - with freezegun.freeze_time("2012-10-11 10:10:10"): - args = ["-c", config, "snapshot", "create"] - main(args) - args = ["-c", config, "publish", "update"] - main(args) - state = SystemStateReader() - state.read() - expect = set( - [ - "archived-fakerepo01-20121011T1010Z", - "fakerepo01-20121011T0000Z", - "fakerepo02-20121006T0000Z", - "fakerepo01-20121010T0000Z", - ] - ) - assert expect == state.snapshots - expect = { - "fakerepo02 main": set(["fakerepo02-20121006T0000Z"]), - "fakerepo01 main": set(["fakerepo01-20121011T0000Z"]), - } - assert expect == state.publish_map - - -def do_repo_create(config): # pragma: no cover - """Test if creating repositories works.""" - args = ["-c", config, "repo", "create"] - main(args) - state = SystemStateReader() - state.read() - call_output( - [ - "aptly", - "repo", - "add", - "centrify", - "/source/compose/setup/hellome_0.1-1_amd64.deb", - ] - ) - assert set(["centrify"]) == state.repos - - -@pytest.mark.skip -def test_repo_create_single(): - """Test if creating a single repo works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"repo.yml", - ) - ) as (tyml, config): - args = [ - "-c", - config, - "repo", - "create", - "centrify", - ] - main(args) - state = SystemStateReader() - state.read() - assert set(["centrify"]) == state.repos - - -@pytest.mark.skip -def test_repo_create_inexistent(): - """Test if creating an inexistent repo causes an error.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"repo.yml", - ) - ) as (tyml, config): - args = [ - "-c", - config, - "repo", - "create", - "asdfasdf", - ] - error = False - try: - main(args) - except ValueError: - error = True - assert error - - -@pytest.mark.skip -def test_repo_create_basic(): - """Test if creating repositories works.""" - with test.clean_and_config( - os.path.join( - _test_base, - b"repo.yml", - ) - ) as (tyml, config): - do_repo_create(config) - - -@pytest.mark.skip -def test_snapshot_spec_as_dict(): - "Test various snapshot formats for snapshot_spec_to_name()" - - snap_string = "snapshot-foo" - snap_dict = {"name": "foo"} - - cfg = { - "snapshot": { - "foo": {}, - } - } - - assert snapshot_spec_to_name(cfg, snap_string) == snap_string - assert snapshot_spec_to_name(cfg, snap_dict) == "foo" From 135ac740caa6ec59ce33e398dc89abcfdcc865c9 Mon Sep 17 00:00:00 2001 From: Jean-Louis Fuchs Date: Tue, 16 Jan 2024 17:14:41 +0100 Subject: [PATCH 3/3] test: move hypthesis and support custom keyservers --- pyaptly/conftest.py | 2 + pyaptly/legacy.py | 7 ++- pyaptly/test.py | 4 +- pyaptly/tests/debug.toml | 38 ++++++++++++ .../{graph_test.py => tests/test_graph.py} | 58 ++++++++----------- pyaptly/tests/test_mirror.py | 16 +++++ pyaptly/tests/test_util.py | 5 ++ pyaptly/util.py | 13 ++++- 8 files changed, 104 insertions(+), 39 deletions(-) create mode 100644 pyaptly/tests/debug.toml rename pyaptly/{graph_test.py => tests/test_graph.py} (72%) diff --git a/pyaptly/conftest.py b/pyaptly/conftest.py index 6cea7b2..ec66db3 100644 --- a/pyaptly/conftest.py +++ b/pyaptly/conftest.py @@ -66,10 +66,12 @@ def environment(debug_mode): os.chown(gnupg, 0, 0) gnupg.chmod(0o700) os.environ["GNUPGHOME"] = str(gnupg) + util._PYTEST_KEYSERVER = "hkp://127.0.0.1:8080" try: yield finally: + util._PYTEST_KEYSERVER = None tempdir_obj.cleanup() aptly_conf.unlink() diff --git a/pyaptly/legacy.py b/pyaptly/legacy.py index 2e4bdec..ae58487 100755 --- a/pyaptly/legacy.py +++ b/pyaptly/legacy.py @@ -17,6 +17,8 @@ import six import yaml +from .util import get_default_keyserver + _logging_setup = False if six.PY2: @@ -1562,6 +1564,9 @@ def add_gpg_keys(mirror_config): :param mirror_config: The configuration yml as dict :type mirror_config: dict """ + keyserver = mirror_config.get("keyserver") + if not keyserver: + keyserver = get_default_keyserver() keys_urls = {} if "gpg-keys" in mirror_config: keys = unit_or_list_to_list(mirror_config["gpg-keys"]) @@ -1588,7 +1593,7 @@ def add_gpg_keys(mirror_config): "--keyring", "trustedkeys.gpg", "--keyserver", - "hkp://127.0.0.1:8080", + keyserver, "--recv-keys", key, ] diff --git a/pyaptly/test.py b/pyaptly/test.py index 78f0ada..4e62c42 100644 --- a/pyaptly/test.py +++ b/pyaptly/test.py @@ -73,7 +73,7 @@ def merge(a, b): return b -def execute_and_parse_show_cmd(args): +def execute_and_parse_show_cmd(args): # pragma: no cover """Executes and parses a aptly show command. :param args: Command to execute @@ -105,7 +105,7 @@ def create_config(test_input): mirror["components"] = "main" if "distribution" not in mirror: mirror["distribution"] = "main" - if "publish" in input_: + if "publish" in input_: # pragma: no cover for publish in input_["publish"].values(): for item in publish: if "components" not in item: diff --git a/pyaptly/tests/debug.toml b/pyaptly/tests/debug.toml new file mode 100644 index 0000000..b85931b --- /dev/null +++ b/pyaptly/tests/debug.toml @@ -0,0 +1,38 @@ +[mirror.trusty] +components = [ "main", "multiverse", "restricted", "universe",] +architectures = [ "amd64", "i386",] +distribution = "trusty" +sources = true +udeb = true +archive = "http://ch.archive.ubuntu.com/ubuntu/" +gpg-keys = [ "40976EAF437D05B5", "3B4FE6ACC0B21F32" ] +keyserver = "keyserver.ubuntu.com" + +[mirror.trusty-updates] +components = [ "main", "multiverse", "restricted", "universe",] +architectures = [ "amd64", "i386",] +distribution = "trusty-updates" +sources = true +udeb = true +archive = "http://ch.archive.ubuntu.com/ubuntu/" +gpg-keys = [ "40976EAF437D05B5", "3B4FE6ACC0B21F32" ] +keyserver = "keyserver.ubuntu.com" + +[mirror.trusty-backports] +components = [ "main", "multiverse", "restricted", "universe",] +architectures = [ "amd64", "i386",] +distribution = "trusty-backports" +sources = true +udeb = true +archive = "http://ch.archive.ubuntu.com/ubuntu/" +gpg-keys = [ "40976EAF437D05B5", "3B4FE6ACC0B21F32" ] +keyserver = "keyserver.ubuntu.com" + +[snapshot.trusty-latest] +mirror = "trusty-updates" + +[snapshot.trusty-backports-latest] +mirror = "trusty-backports" + +[snapshot.my-repo-latest] +repo = "my-repo" diff --git a/pyaptly/graph_test.py b/pyaptly/tests/test_graph.py similarity index 72% rename from pyaptly/graph_test.py rename to pyaptly/tests/test_graph.py index 2f08255..0589c15 100644 --- a/pyaptly/graph_test.py +++ b/pyaptly/tests/test_graph.py @@ -1,38 +1,24 @@ -# type: ignore # TODO -# flake8: noqa # TODO - -"""Testing dependency graphs""" +"""Testing dependency graphs.""" import random -import sys - -from pyaptly.legacy import Command, FunctionCommand +from functools import partial -from . import test +from hypothesis import given, settings +from hypothesis import strategies as st -if not sys.version_info < (2, 7): # pragma: no cover - from hypothesis import strategies as st - from hypothesis import given - -from hypothesis import settings +from pyaptly.legacy import Command, FunctionCommand # type: ignore # Disable the deadline globally for all tests settings.register_profile("my_profile", deadline=None) settings.load_profile("my_profile") -if sys.version_info < (2, 7): # pragma: no cover - import mock - - given = mock.MagicMock() # noqa - example = mock.MagicMock() # noqa - st = mock.MagicMock() # noqa - RES_COUNT = 20 range_intagers_st = st.integers(min_value=0, max_value=RES_COUNT) @st.composite -def provide_require_st(draw, filter_=True): # pragma: no cover +def provide_require_st(draw, filter_=True): + """Build a random command tree, to test.""" commands = draw(range_intagers_st) provides = draw( st.lists( @@ -53,7 +39,7 @@ def provide_require_st(draw, filter_=True): # pragma: no cover else: max_prov = 0 if filter_: - provides_filter = [x for x in provides_set if x > max_prov] + provides_filter = set([x for x in provides_set if x > max_prov]) else: provides_filter = provides_set if provides_filter: @@ -67,6 +53,7 @@ def provide_require_st(draw, filter_=True): # pragma: no cover def print_example(): # pragma: no cover + """Print an example for debugging.""" example = provide_require_st().example() print( """ @@ -85,17 +72,17 @@ def print_example(): # pragma: no cover print("}") -@test.hypothesis_min_ver @given(provide_require_st(), st.random_module()) -def test_graph_basic(tree, rnd): # pragma: no cover - """Test our test method, create a basic graph using hypthesis and run some - basic tests against it.""" +def test_graph_basic(tree, rnd): + """Test our test method, create a basic graph using hypthesis. + + And run some basic tests against it. + """ run_graph(tree) -@test.hypothesis_min_ver @given(provide_require_st(False), st.random_module()) -def test_graph_cycles(tree, rnd): # pragma: no cover +def test_graph_cycles(tree, rnd): """Test reacts correctly on trees with cycles.""" try: run_graph(tree) @@ -104,26 +91,27 @@ def test_graph_cycles(tree, rnd): # pragma: no cover raise e -@test.hypothesis_min_ver @given(provide_require_st(), provide_require_st(), st.random_module()) def test_graph_island(tree0, tree1, rnd): # pragma: no cover - """Test with two independant graphs which can form a island""" + """Test with two independant graphs which can form a island.""" tree = (tree0[0] + tree1[0], tree0[1] + tree1[1], tree0[2] + tree1[2]) run_graph(tree) -def run_graph(tree): # pragma: no cover - """Runs the test""" +def run_graph(tree): + """Run the test.""" commands = [] index = list(range(len(tree[0]))) random.shuffle(index) for i in index: - def dummy(): # pragma: no cover + def dummy(i): # pragma: no cover return i if tree[2][i]: - cmd = FunctionCommand(dummy) + func = partial(dummy, i) + func.__name__ = dummy.__name__ # type: ignore + cmd = FunctionCommand(func) else: cmd = Command(i) for provides in tree[0][i]: @@ -133,7 +121,7 @@ def dummy(): # pragma: no cover commands.append(cmd) ordered = Command.order_commands(commands) assert len(commands) == len(ordered) - provided = set() + provided: set[tuple[str, str]] = set() for command in ordered: assert command._requires.issubset(provided) provided.update(command._provides) diff --git a/pyaptly/tests/test_mirror.py b/pyaptly/tests/test_mirror.py index 6cbb939..a90ddf9 100644 --- a/pyaptly/tests/test_mirror.py +++ b/pyaptly/tests/test_mirror.py @@ -1,9 +1,25 @@ """Test mirror functionality.""" +import logging + import pytest import pyaptly +@pytest.mark.parametrize("config", ["debug.toml"], indirect=True) +def test_debug(environment, config): + """Test if debug is enabled with -d.""" + args = [ + "-d", + "-c", + config, + "mirror", + "create", + ] + pyaptly.main(args) + assert logging.getLogger().level == logging.DEBUG + + @pytest.mark.parametrize("config", ["mirror-extra.toml"], indirect=True) def test_mirror_create(environment, config, caplog): """Test if creating mirrors works.""" diff --git a/pyaptly/tests/test_util.py b/pyaptly/tests/test_util.py index 04f9042..6a98b3e 100644 --- a/pyaptly/tests/test_util.py +++ b/pyaptly/tests/test_util.py @@ -55,3 +55,8 @@ def test_snapshot_spec_as_dict(): assert legacy.snapshot_spec_to_name(cfg, snap_string) == snap_string # type: ignore assert legacy.snapshot_spec_to_name(cfg, snap_dict) == "foo" # type: ignore + + +def test_get_default_keyserver(): + """Test getting default keyserver.""" + assert util.get_default_keyserver() == util._DEFAULT_KEYSERVER diff --git a/pyaptly/util.py b/pyaptly/util.py index 32b65ca..7f0888b 100644 --- a/pyaptly/util.py +++ b/pyaptly/util.py @@ -4,7 +4,10 @@ import subprocess from pathlib import Path from subprocess import DEVNULL, PIPE # noqa: F401 -from typing import Union +from typing import Optional, Union + +_DEFAULT_KEYSERVER: str = "hkps://keys.openpgp.org" +_PYTEST_KEYSERVER: Optional[str] = None _DEBUG = False _PYTEST_DEBUG = False @@ -21,6 +24,14 @@ logger = logging.getLogger(__name__) +def get_default_keyserver(): + """Get default keyseerver.""" + if _PYTEST_KEYSERVER: + return _PYTEST_KEYSERVER + else: + return _DEFAULT_KEYSERVER + + def is_debug_mode(): """Check if we are in debug mode.""" return _DEBUG or _PYTEST_DEBUG