diff --git a/tests/test_ubipop.py b/tests/test_ubipop.py index 346fd94..48a2778 100644 --- a/tests/test_ubipop.py +++ b/tests/test_ubipop.py @@ -10,7 +10,7 @@ import pytest import ubiconfig -from mock import MagicMock, patch, call +from mock import MagicMock, patch, call, ANY from more_executors import Executors from ubipop import UbiPopulateRunner, UbiRepoSet, RepoSet, UbiPopulate from ubipop._pulp_client import Module, ModuleDefaults, Package, Repo @@ -851,6 +851,75 @@ def test_create_output_file_all_repos(mock_ubipop_runner, mock_get_repo_pairs, shutil.rmtree(path) +@patch("ubipop.UbiPopulateRunner._get_current_content") +@patch("ubipop._pulp_client.Pulp.search_repo_by_cs") +def test_expected_pulp_actions(mocked_search_repo_by_cs, mocked_current_content, + mock_current_content_ft): + # Don't actually query Pulp for repos or their content + mocked_search_repo_by_cs.side_effect = [ + # Input repos - rhel-7-server + [get_test_repo( + repo_id="rhel-7-server-rpms__7_DOT_2__x86_64", + content_set="rhel-7-server-rpms", + ), ], + [get_test_repo( + repo_id="rhel-7-server-source-rpms__7_DOT_2__x86_64", + content_set="rhel-7-server-source-rpms", + ), ], + [get_test_repo( + repo_id="rhel-7-server-debuginfo-rpms__7_DOT_2__x86_64", + content_set="rhel-7-server-debuginfo-rpms", + ), ], + + # Output repos - rhel-7-server + [get_test_repo( + repo_id="ubi-7-server-rpms__7_DOT_2__x86_64", + content_set="ubi-7-server-rpms", + ubi_population=True + ), ], + [get_test_repo( + repo_id="ubi-7-server-source-rpms__7_DOT_2__x86_64", + content_set="ubi-7-server-source-rpms", + ubi_population=True + ), ], + [get_test_repo( + repo_id="ubi-7-server-debuginfo-rpms__7_DOT_2__x86_64", + content_set="ubi-7-server-debuginfo-rpms", + ubi_population=True + ), ], + ] + mocked_current_content.return_value = mock_current_content_ft + + actions = UbiPopulate( + "foo.pulp.com", + ('foo', 'foo'), + False, + ubiconfig_dir_or_url=TEST_DATA_DIR, + ).expected_pulp_actions(content_sets=["rhel-7-server-rpms", ]) + + # associations unit_id, type, src_repo, dest_repo + assert len(actions.associations) == 0 + + # unassociations unit_id, type, dest_repo + assert len(actions.unassociations) == 4 + + assert actions.unassociations[0].unit_id == "md_current" + assert actions.unassociations[0].type == "modules" + assert actions.unassociations[0].dest_repo == "ubi-7-server-rpms__7_DOT_2__x86_64" + + assert actions.unassociations[1].unit_id == "rpm_current" + assert actions.unassociations[1].type == "packages" + assert actions.unassociations[1].dest_repo == "ubi-7-server-rpms__7_DOT_2__x86_64" + + assert actions.unassociations[2].unit_id == "srpm_current" + assert actions.unassociations[2].type == "packages" + assert actions.unassociations[2].dest_repo == "ubi-7-server-source-rpms__7_DOT_2__x86_64" + + assert actions.unassociations[3].unit_id == "debug_rpm_current" + assert actions.unassociations[3].type == "packages" + assert actions.unassociations[3].dest_repo == "ubi-7-server-debuginfo-rpms__7_DOT_2__x86_64" + + @pytest.fixture(name='mock_current_content_ft') def fixture_mock_current_content_ft(): current_modules_ft = MagicMock() diff --git a/ubipop/__init__.py b/ubipop/__init__.py index 4d3582e..9a53916 100644 --- a/ubipop/__init__.py +++ b/ubipop/__init__.py @@ -27,6 +27,7 @@ class RepoMissing(Exception): RepoSet = namedtuple('RepoSet', ['rpm', 'source', 'debug']) +ExpectedPulpActions = namedtuple('ExpectedPulpActions', ['associations', 'unassociations']) class UbiRepoSet(object): @@ -132,6 +133,25 @@ def _filter_ubi_conf_list(self, config_list, content_sets, repo_ids): return filtered_conf_list + def expected_pulp_actions(self, content_sets=None, repo_ids=None): + expected_pulp_actions = ExpectedPulpActions([], []) + + for config in self._filter_ubi_conf_list(self.ubiconfig_list, content_sets, repo_ids): + try: + repo_pairs = self._get_input_and_output_repo_pairs(config) + except RepoMissing: + _LOG.warning("Skipping current content triplet, some repos are missing") + continue + + for repo_set in repo_pairs: + run = UbiPopulateRunner(self.pulp, repo_set, config, self.dry_run, self._executor) + ubi_actions = run.expected_pulp_actions() + + expected_pulp_actions.associations.extend(ubi_actions.associations) + expected_pulp_actions.unassociations.extend(ubi_actions.unassociations) + + return expected_pulp_actions + def populate_ubi_repos(self): out_repos = set() @@ -505,6 +525,28 @@ def _diff_lists_by_attr(self, list_1, list_2, attr): return diff + def expected_pulp_actions(self): + actions = ExpectedPulpActions([], []) + + Association = namedtuple('Association', ['unit_id', 'type', 'src_repo', 'dest_repo']) + Unassociation = namedtuple('Unassociation', ['unit_id', 'type', 'dest_repo']) + + associations, unassociations, _, _ = self._get_pulp_actions(*self._get_current_content()) + + for item in associations: + if item.units: + for unit in item.units: + actions.associations.append(Association(unit.name, item.TYPE, item.src_repo.repo_id, + item.dst_repo.repo_id)) + + for item in unassociations: + if item.units: + for unit in item.units: + actions.unassociations.append(Unassociation(unit.name, item.TYPE, + item.dst_repo.repo_id)) + + return actions + def run_ubi_population(self): current_modules_ft, current_module_defaults_ft, current_rpms_ft, \ current_srpms_ft, current_debug_rpms_ft = self._get_current_content()