Extend the test case by real changes in publications at Caps side

tests/foreman/cli/test_capsulecontent.py

@@ -352,25 +352,33 @@ def test_positive_content_counts_granularity(
 
     :parametrized: yes
 
+    :verifies: SAT-28337
+
     :setup:
         1. Satellite with registered external Capsule.
 
     :steps:
         1. Create two LCEs, assign them to the Capsule.
         2. Create and sync a repo, publish it in two CVs, promote both CVs to both LCEs.
         3. Ensure full counts were calculated for both CVs, both LCEs.
-        4. Create a filter for both CVs to change the counts of next version.
-        5. Publish new version of the first CV, promote it to the first LCE.
-        6. Ensure counts of the first CV were updated in the first LCE only, nothing else.
-        7. Promote the first CV to the second LCE.
-        8. Ensure counts for the first CV were updated in the second LCE, no changes for second CV.
+        4. Disable automatic content counts update.
+        5. Create a filter for both CVs to change the counts of next version.
+        6. Publish new version of both CVs and promote to both LCEs.
+           After capsule sync completes we should have changes in all publictions.
+        7. Ensure the counts were invalidated in both CVs, both LCEs.
+        8. Enable automatic content counts update.
+        9. Publish new version of the first CV, promote it to the first LCE.
+        10. Ensure counts of the first CV were updated in the first LCE only, nothing else.
+        11. Promote the first CV to the second LCE.
+        12. Ensure counts for the first CV were updated in the second LCE, no changes for second CV.
 
     :expectedresults:
         1. Content counts are updated only for particular CVs promoted to particular LCEs.
     """
     wait_query = (
         'label = Actions::Katello::ContentView::CapsuleSync OR '
-        'Actions::Katello::CapsuleContent::UpdateContentCounts'
+        'Actions::Katello::CapsuleContent::UpdateContentCounts '
+        'AND started_at >= "{}"'
     )
 
     # Create two LCEs, assign them to the Capsule.
@@ -398,7 +406,11 @@ def test_positive_content_counts_granularity(
         cv.version[0].promote(data={'environment_ids': [lce1.id, lce2.id]})
         cvs.append(cv.read())
     cv1, cv2 = cvs
-    module_target_sat.wait_for_tasks(search_query=wait_query, search_rate=5, max_tries=5)
+    module_target_sat.wait_for_tasks(
+        search_query=wait_query.format(datetime.utcnow().replace(microsecond=0)),
+        search_rate=5,
+        max_tries=5,
+    )
 
     # Ensure full counts were calculated for both CVs, both LCEs.
     info = module_target_sat.cli.Capsule.content_info(
@@ -417,6 +429,10 @@ def test_positive_content_counts_granularity(
                 int(info[lce.name][cv.name][repo.name]['errata']) == repo.content_counts['erratum']
             )
 
+    # Disable automatic content counts update.
+    setting_update.value = False
+    setting_update = setting_update.update({'value'})
+
     # Create a filter for both CVs to change the counts of next version.
     for cv in cvs:
         cvf = module_target_sat.cli_factory.make_content_view_filter(
@@ -433,12 +449,40 @@ def test_positive_content_counts_granularity(
             }
         )
 
+    # Publish new version of both CVs and promote to both LCEs.
+    # After capsule sync completes we should have changes in all publictions.
+    for cv in cvs:
+        cv.publish()
+        cv = cv.read()
+        cv.version[0].promote(data={'environment_ids': [lce1.id, lce2.id]})
+        cv = cv.read()
+    module_target_sat.wait_for_tasks(
+        search_query=wait_query.format(datetime.utcnow().replace(microsecond=0)),
+        search_rate=5,
+        max_tries=5,
+    )
+
+    # Ensure the counts were invalidated in both CVs, both LCEs.
+    info = module_target_sat.cli.Capsule.content_info(
+        {'id': module_capsule_configured.nailgun_capsule.id, 'organization-id': function_org.id}
+    )
+    info = _dictionarize_counts(info)
+    assert all(info[lce.name][cv.name][repo.name] == {} for cv in cvs for lce in [lce1, lce2])
+
+    # Enable automatic content counts update.
+    setting_update.value = True
+    setting_update = setting_update.update({'value'})
+
     # Publish new version of the first CV, promote it to the first LCE.
     cv1.publish()
     cv1 = cv1.read()
     cv1.version[0].promote(data={'environment_ids': lce1.id})
     cv1 = cv1.read()
-    module_target_sat.wait_for_tasks(search_query=wait_query, search_rate=5, max_tries=5)
+    module_target_sat.wait_for_tasks(
+        search_query=wait_query.format(datetime.utcnow().replace(microsecond=0)),
+        search_rate=5,
+        max_tries=5,
+    )
 
     # Ensure counts of the first CV were updated in the first LCE only, nothing else.
     info = module_target_sat.cli.Capsule.content_info(
@@ -449,24 +493,21 @@ def test_positive_content_counts_granularity(
     assert int(info[lce1.name][cv1.name][repo.name]['packages']) == 1
     assert int(info[lce1.name][cv1.name][repo.name]['package-groups']) == 1
     assert int(info[lce1.name][cv1.name][repo.name]['errata']) == 1
-
-    for lce in [lce1, lce2]:
-        for cv in cvs:
-            if cv.name == cv1.name and lce.name == lce1.name:
-                continue
-            assert int(info[lce.name][cv.name][repo.name]['packages']) == repo.content_counts['rpm']
-            assert (
-                int(info[lce.name][cv.name][repo.name]['package-groups'])
-                == repo.content_counts['package_group']
-            )
-            assert (
-                int(info[lce.name][cv.name][repo.name]['errata']) == repo.content_counts['erratum']
-            )
+    assert all(
+        info[lce.name][cv.name][repo.name] == {}
+        for cv in cvs
+        for lce in [lce1, lce2]
+        if (cv.name != cv1.name or lce.name != lce1.name)
+    )
 
     # Promote the first CV to the second LCE.
     cv1.version[0].promote(data={'environment_ids': lce2.id})
     cv1 = cv1.read()
-    module_target_sat.wait_for_tasks(search_query=wait_query, search_rate=5, max_tries=5)
+    module_target_sat.wait_for_tasks(
+        search_query=wait_query.format(datetime.utcnow().replace(microsecond=0)),
+        search_rate=5,
+        max_tries=5,
+    )
 
     # Ensure counts for the first CV were updated in the second LCE, no changes for second CV.
     info = module_target_sat.cli.Capsule.content_info(
@@ -477,14 +518,7 @@ def test_positive_content_counts_granularity(
     assert int(info[lce2.name][cv1.name][repo.name]['packages']) == 1
     assert int(info[lce2.name][cv1.name][repo.name]['package-groups']) == 1
     assert int(info[lce2.name][cv1.name][repo.name]['errata']) == 1
-
-    for lce in [lce1, lce2]:
-        assert int(info[lce.name][cv2.name][repo.name]['packages']) == repo.content_counts['rpm']
-        assert (
-            int(info[lce.name][cv2.name][repo.name]['package-groups'])
-            == repo.content_counts['package_group']
-        )
-        assert int(info[lce.name][cv2.name][repo.name]['errata']) == repo.content_counts['erratum']
+    assert all(info[lce.name][cv2.name][repo.name] == {} for lce in [lce1, lce2])
 
 
 @pytest.mark.tier4