forked from SatelliteQE/robottelo
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_errata.py
More file actions
1660 lines (1479 loc) · 65.4 KB
/
Copy pathtest_errata.py
File metadata and controls
1660 lines (1479 loc) · 65.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""UI Tests for the errata management feature
:Requirement: Errata
:CaseAutomation: Automated
:CaseComponent: ErrataManagement
:team: Artemis
:CaseImportance: High
"""
from datetime import UTC, datetime, timedelta
import re
from broker import Broker
from dateutil import parser
from dateutil.parser import parse
from fauxfactory import gen_string
import pytest
from selenium.common.exceptions import NoSuchElementException
from wait_for import wait_for
from robottelo import constants
from robottelo.config import settings
from robottelo.constants import (
DEFAULT_LOC,
FAKE_1_CUSTOM_PACKAGE,
FAKE_1_CUSTOM_PACKAGE_NAME,
FAKE_2_CUSTOM_PACKAGE,
FAKE_3_YUM_OUTDATED_PACKAGES,
FAKE_4_CUSTOM_PACKAGE,
FAKE_5_CUSTOM_PACKAGE,
FAKE_9_YUM_OUTDATED_PACKAGES,
FAKE_9_YUM_SECURITY_ERRATUM,
FAKE_9_YUM_SECURITY_ERRATUM_COUNT,
FAKE_10_YUM_BUGFIX_ERRATUM,
FAKE_10_YUM_BUGFIX_ERRATUM_COUNT,
FAKE_11_YUM_ENHANCEMENT_ERRATUM,
FAKE_11_YUM_ENHANCEMENT_ERRATUM_COUNT,
PRDS,
REAL_0_RH_PACKAGE,
REAL_4_ERRATA_CVES,
REAL_4_ERRATA_ID,
REAL_RHEL8_1_ERRATA_ID,
REAL_RHEL8_ERRATA_CVES,
REAL_RHSCLIENT_ERRATA,
TIMESTAMP_FMT,
)
from robottelo.hosts import ContentHost
from robottelo.utils.issue_handlers import is_open
CUSTOM_REPO_URL = settings.repos.yum_9.url
CUSTOM_REPO_ERRATA = settings.repos.yum_9.errata
CUSTOM_REPO_ERRATA_ID = settings.repos.yum_9.errata[0]
CUSTOM_REPO_3_URL = settings.repos.yum_3.url
CUSTOM_REPO_3_ERRATA = settings.repos.yum_3.errata
CUSTOM_REPO_3_ERRATA_ID = settings.repos.yum_3.errata[0]
RHVA_PACKAGE = REAL_0_RH_PACKAGE
RHVA_ERRATA_ID = REAL_4_ERRATA_ID
RHVA_ERRATA_CVES = REAL_4_ERRATA_CVES
pytestmark = [pytest.mark.run_in_one_thread]
def _generate_errata_applicability(hostname, module_target_sat):
"""Force host to generate errata applicability"""
host = module_target_sat.api.Host().search(query={'search': f'name={hostname}'})[0].read()
host.errata_applicability(synchronous=False)
def _set_setting_value(setting_entity, value):
"""Set setting value.
:param setting_entity: the setting entity instance.
:param value: The setting value to set.
"""
setting_entity.value = value
setting_entity.update(['value'])
@pytest.fixture
def errata_status_installable(module_target_sat):
"""Fixture to allow restoring errata_status_installable setting after usage"""
errata_status_installable = module_target_sat.api.Setting().search(
query={'search': 'name="errata_status_installable"'}
)[0]
original_value = errata_status_installable.value
yield errata_status_installable
_set_setting_value(errata_status_installable, original_value)
def cv_publish_promote(sat, org, cv, lce=None, needs_publish=True):
"""Publish & promote Content View Version with all content visible in org.
:param lce: if None, default to 'Library',
pass a single instance of lce, or list of instances.
do not pass the Library environment.
:param bool needs_publish: if False, skip publish of a new version
:return dictionary:
'content-view': instance of updated cv
'content-view-version': instance of newest cv version
"""
# Default to 'Library' lce, if None passed
# Take a single instance of lce, or list of instances
lce_ids = 'Library'
if lce is not None:
lce_ids = [lce.id] if not isinstance(lce, list) else sorted(_lce.id for _lce in lce)
if needs_publish is True:
_publish_and_wait(sat, org, cv)
# Content-view must have at least one published version
cv = sat.api.ContentView(id=cv.id).read()
assert cv.version, f'No version(s) are published to the Content-View: {cv.id}'
# Find highest version id, will be the latest
cvv_id = max(cvv.id for cvv in cv.version)
# Promote to lifecycle-environment(s)
if lce_ids == 'Library':
library_lce = cv.environment[0].read()
sat.api.ContentViewVersion(id=cvv_id).promote(
data={'environment_ids': library_lce.id, 'force': 'True'}
)
else:
sat.api.ContentViewVersion(id=cvv_id).promote(data={'environment_ids': lce_ids})
_result = {
'content-view': sat.api.ContentView(id=cv.id).read(),
'content-view-version': sat.api.ContentViewVersion(id=cvv_id).read(),
}
assert all(entry for entry in _result.values()), (
f'One or more necessary components are missing: {_result}'
)
return _result
def _publish_and_wait(sat, org, cv, timeout=60):
"""Synchrnous publish of a new version of content-view to organization,
wait for task completion.
return: the polled task, success or fail.
"""
task_id = sat.api.ContentView(id=cv.id).publish({'id': cv.id, 'organization': org})['id']
assert task_id, f'No task was invoked to publish the Content-View: {cv.id}.'
# Should take < 1 minute, check in 5s intervals
sat.wait_for_tasks(
search_query=(f'label = Actions::Katello::ContentView::Publish and id = {task_id}'),
search_rate=5,
max_tries=round(timeout / 5),
)
return sat.api.ForemanTask(id=task_id).poll(must_succeed=False)
@pytest.fixture
def registered_contenthost(
module_sca_manifest_org,
module_target_sat,
rhel_contenthost,
module_product,
module_lce,
module_ak,
module_cv,
request,
repos=None,
):
"""RHEL ContentHost registered in satellite,
Using SCA and global registration.
:note: rhel_contenthost will be parametrized by rhel6 to 9, also -fips for all distros.
to use specific rhel version parametrized contenthost;
use `pytest.mark.rhel_ver_match('[]')` to mark contenthost rhel major version(s)
for tests using this fixture.
:environment: Defaults to module_lce.
To use Library environment for activation key / content-view:
pass the string 'Library' (not case sensitive) in the list of params.
:repos: pass as a parametrized request
list of upstream URLs for custom repositories.
default: None; repo enablement will be sklipped for host.
if None, add any repos to cv/ak, publish/promote etc, after calling fixture.
example:
@pytest.mark.parametrize('registered_contenthost',
[[repo1_url, repo2_url,]],
indirect=True,
)
for Library env:
@pytest.mark.parametrize('registered_contenthost',
[['library', repo1_url, repo2_url,]],
indirect=True,
)
for Default: no repos, use module_cv, module_ak, module_lce:
no need to parametrize fixture, just import it.
if desired, still parametrize registered host's rhel major version(s):
eg. pytest.mark.rhel_ver_match('[8, 9, ...]') etc.
"""
params = getattr(request, 'param', None)
environment = module_lce
if params is None:
repos = []
else:
if any(p.lower() == 'library' for p in params):
environment = 'Library'
repos = [p for p in params if str(p).lower() != 'library']
if rhel_contenthost.subscribed:
rhel_contenthost.unregister()
custom_repos = []
for repo_url in repos:
new_repo = module_target_sat.api.Repository(url=repo_url, product=module_product).create()
new_repo.sync()
custom_repos.append(new_repo)
if len(custom_repos) > 0:
module_cv.repository = custom_repos
module_cv.update(['repository'])
# Publish/promote CV if needed, associate entities, register client:
# skip enabling repos, we will do after, with subscription-manager
setup = module_target_sat.api_factory.register_host_and_needed_setup(
organization=module_sca_manifest_org,
client=rhel_contenthost,
activation_key=module_ak,
environment=environment,
content_view=module_cv,
)
@request.addfinalizer
# Cleanup for in-between parametrized sessions,
# unregister the host if it's still subscribed to content.
def cleanup():
nonlocal setup
client = setup['client']
if client and client.subscribed:
client.unregister()
# no error setting up fixtures and registering client
assert setup['result'] != 'error', f'{setup["message"]}'
assert (client := setup['client'])
# nothing applicable to start
result = client.execute('subscription-manager repos')
assert client.applicable_errata_count == 0
assert client.applicable_package_count == 0
# if no repos given, subscription-manager should report error
if len(repos) == 0:
assert client.execute(r'subscription-manager repos --enable \*').status == 1
# any custom repos in host are setup, and can be synced again,
# we can also enable each repo with subscription-manager:
else:
# list all repos available to sub-manager:
sub_manager_repos = client.execute('subscription-manager repos --list')
repo_ids_names = {'ids': [], 'names': []}
for line in sub_manager_repos.stdout.splitlines():
# in each output line, check for Name: and ID: of repos listed
if search := re.search(r'ID: (.*)', line):
id_found = search.group(1).strip()
repo_ids_names['ids'].append(id_found)
if search := re.search(r'Name: (.*)', line):
name_found = search.group(1).strip()
repo_ids_names['names'].append(name_found)
# every repo id found, has a corresponding name we will match to satellite repo
assert len(repo_ids_names['ids']) == len(repo_ids_names['names']), (
f"Failed to extract a given repository's name, and or id, from subscription-manager list."
f" {sub_manager_repos}"
)
for repo in custom_repos:
# sync repository to satellite
assert module_target_sat.api.Repository(id=repo.id).read()
result = repo.sync()['humanized']
assert len(result['errors']) == 0, (
f'Failed to sync custom repository [id: {repo.id}]:\n{str(result["errors"])}'
)
# found index (repo) with matching name, grab sub-manager repo-id:
assert repo.name in repo_ids_names['names']
sub_man_repo_id = repo_ids_names['ids'][repo_ids_names['names'].index(repo.name)]
# repo can be enabled by id without error
enable_repo = client.execute(f'subscription-manager repos --enable {sub_man_repo_id}')
assert enable_repo.status == 0, (
f'Failed to enable a repository with subscription-manager, on client: {client.hostname}.'
f' {enable_repo.stderr}'
)
assert len(custom_repos) == len(repo_ids_names['ids'])
assert all(name in repo_ids_names['names'] for r in custom_repos for name in [r.name])
return client
@pytest.mark.e2e
@pytest.mark.rhel_ver_match('N-3') # Newest major RHEL version (N), and three prior.
@pytest.mark.parametrize('registered_contenthost', [[CUSTOM_REPO_URL]], indirect=True)
@pytest.mark.no_containers
def test_end_to_end(
registered_contenthost,
module_target_sat,
module_product,
module_lce,
module_cv,
session,
):
"""Create all entities required for errata, register an applicable host,
read errata details and apply it to host.
:id: a26182fc-f31a-493f-b094-3f5f8d2ece47
:setup: A host with content from a custom repo,
contains some outdated packages applicable errata.
:expectedresults: Errata details are the same as expected, errata
installation is successful.
:parametrized: yes
:Verifies: SAT-23414, SAT-7998
:customerscenario: true
"""
ERRATA_DETAILS = {
'advisory': 'RHSA-2012:0055',
'cves': 'N/A',
'type': 'Security Advisory',
'severity': 'N/A',
'reboot_suggested': 'No',
'topic': '',
'description': 'Sea_Erratum',
'issued': '2012-01-27',
'last_updated_on': '2012-01-27',
'solution': '',
}
ERRATA_PACKAGES = {
'independent_packages': [
'penguin-0.9.1-1.noarch',
'shark-0.1-1.noarch',
'walrus-5.21-1.noarch',
],
'module_stream_packages': [],
}
# client was registered with single custom repo
client = registered_contenthost
hostname = client.hostname
assert client.subscribed
custom_repo = module_cv.read().repository[0].read()
# nothing applicable to start
assert 0 == client.applicable_errata_count == client.applicable_package_count, (
f'Expected no applicable erratum or packages to start, on host: {hostname}'
)
# install outdated package version, making an errata applicable
result = client.execute(f'yum install -y {FAKE_1_CUSTOM_PACKAGE}')
assert result.status == 0, (
f'Failed to install package {FAKE_1_CUSTOM_PACKAGE}.\n{result.stdout}'
)
# recalculate and assert app errata, after installing outdated pkg
assert client.execute('subscription-manager repos').status == 0
applicable_errata = client.applicable_errata_count
assert applicable_errata == 1, (
f'Expected 1 applicable errata: {CUSTOM_REPO_ERRATA_ID}, after setup. Got {applicable_errata}'
)
with session:
# keep timestamp as datetime obj, so we can subtract later.
# timezone-aware (UTC), as task's start/end times are also timezone-aware.
timestamp_start = datetime.now(UTC).replace(microsecond=0)
# Check selection box function for BZ#1688636
session.location.select(loc_name=DEFAULT_LOC)
results = session.errata.search_content_hosts(
entity_name=CUSTOM_REPO_ERRATA_ID,
value=hostname,
environment=module_lce.name,
)
assert len(results) == 1
# BZ 2265095: Check default columns in table of applicable host:
# from ContentTypes > Errata > Details > Content Hosts tab
assert results[0]['Name'] == hostname
if not is_open('SAT-23414'):
assert str(client.deploy_rhel_version) in results[0]['OS']
assert results[0]['Environment'] == module_lce.name
assert results[0]['Content View'] == module_cv.name
# Check errata details
errata = session.errata.read(CUSTOM_REPO_ERRATA_ID)
assert errata['repositories']['table'], (
f'There are no repositories listed for errata ({CUSTOM_REPO_ERRATA_ID}),',
f' expected to find at least one repository, name: {custom_repo.name}.',
)
# repo/product entry in table match expected
# find the first table entry with the custom repository's name
errata_repo = next(
(
repo
for repo in errata['repositories']['table']
if 'Name' in repo and repo['Name'] == custom_repo.name
),
None,
)
# assert custom repo found and product name
assert errata_repo, (
f'Could not find the errata repository in UI by name: {custom_repo.name}.'
)
assert errata_repo['Name'] == custom_repo.name
assert errata_repo['Product'] == module_product.name, (
'The product name for the errata repository in UI does not match.'
)
# Check all tabs of Errata Details page
assert not ERRATA_DETAILS.items() - errata['details'].items(), (
'Errata details do not match expected values.'
)
assert parse(errata['details']['issued']) == parse(ERRATA_DETAILS['issued']), (
'Errata issued date in UI does not match.'
)
assert parse(errata['details']['last_updated_on']) == parse(
ERRATA_DETAILS['last_updated_on']
), 'Errata last updated date in UI does not match.'
assert set(errata['packages']['independent_packages']) == set(
ERRATA_PACKAGES['independent_packages']
), 'Set of errata packages in UI does not match.'
assert (
errata['packages']['module_stream_packages']
== ERRATA_PACKAGES['module_stream_packages']
), 'Errata module streams in UI does not match.'
# Apply Errata, find REX install task
session.host_new.apply_erratas(
entity_name=hostname,
search=f"errata_id == {CUSTOM_REPO_ERRATA_ID}",
)
# str timestamp to scope tasks, does not include timezone.
install_query = (
f'Install errata {CUSTOM_REPO_ERRATA_ID.lower()} on {hostname}'
f' and started_at >= "{timestamp_start.strftime(TIMESTAMP_FMT)}"'
)
results = module_target_sat.wait_for_tasks(
search_query=install_query,
search_rate=2,
max_tries=60,
)
# should only be one task from this host after timestamp
assert len(results) == 1, (
f'Expected just one errata install task, but found {len(results)}.\nsearch_query: {install_query}'
)
task_status = module_target_sat.api.ForemanTask(id=results[0].id).poll()
assert task_status['result'] == 'success', (
f'Errata Installation task failed:\n{task_status}'
)
assert client.applicable_errata_count == 0, (
f'Unexpected applicable errata found after install of {CUSTOM_REPO_ERRATA_ID}.'
)
# UTC timing for install task and session
install_start = parser.parse(task_status['started_at'])
install_end = parser.parse(task_status['ended_at'])
# install task duration did not exceed 1 minute,
# duration since start of session did not exceed 10 minutes.
assert (install_end - install_start).total_seconds() <= 60
assert (install_end - timestamp_start).total_seconds() <= 600
# Find bulk generate applicability task
results = module_target_sat.wait_for_tasks(
search_query=(f'Bulk generate applicability for host {hostname}'),
search_rate=2,
max_tries=60,
)
results.sort(key=lambda res: res.id)
task_status = module_target_sat.api.ForemanTask(id=results[-1].id).poll()
assert task_status['result'] == 'success', (
f'Bulk Generate Errata Applicability task failed:\n{task_status}'
)
# UTC timing for generate applicability task
bulk_gen_start = parser.parse(task_status['started_at'])
bulk_gen_end = parser.parse(task_status['ended_at'])
assert (bulk_gen_start - install_end).total_seconds() <= 30
assert (bulk_gen_end - bulk_gen_start).total_seconds() <= 60
# Errata should still be visible on satellite, but not on contenthost
assert session.errata.read(CUSTOM_REPO_ERRATA_ID)
results = session.errata.search_content_hosts(
entity_name=CUSTOM_REPO_ERRATA_ID,
value=hostname,
environment=module_lce.name,
)
assert len(results) == 0
# Check package version was updated on contenthost
_package_version = client.execute(f'rpm -q {FAKE_1_CUSTOM_PACKAGE_NAME}').stdout
assert FAKE_2_CUSTOM_PACKAGE in _package_version
@pytest.mark.no_containers
@pytest.mark.rhel_ver_match('N-2')
@pytest.mark.parametrize('registered_contenthost', [[CUSTOM_REPO_3_URL]], indirect=True)
@pytest.mark.skipif((not settings.robottelo.REPOS_HOSTING_URL), reason='Missing repos_hosting_url')
def test_host_content_errata_tab_pagination(
module_sca_manifest_org,
registered_contenthost,
module_target_sat,
module_lce,
module_cv,
session,
):
"""
# Test per-page pagination for BZ#1662254
# Test apply by REX using Select All for BZ#1846670
:setup:
1. registered contenthost with custom repos enabled.
2. enable and sync rh repository.
3. add rh repo to cv for registered host and publish/promote.
:id: 6363eda7-a162-4a4a-b70f-75decbd8202e
:steps:
1. Install more than 20 packages that need errata
2. View Content Host's Errata page
3. Assert total_pages > 1
4. Change per-page setting to 50
5. Assert table has more than 20 errata
6. Change per-page setting to 5
7. Assert setting changed and more total-pages now
8. Search and select one available errata and install it.
9. Assert total items in table is one less.
10. Assert per page count and total pages has not changed.
11. Use the selection box on the left to select all on this page and others.
All errata, from all pages, are selected. Select all YYY.
12. Click the drop down arrow to the right of "Apply All", click Submit.
13. Assert All Errata were applied, none are available anymore.
14. Assert no pagination on new host UI>content>errata.
15. Raise `NoSuchElementException` when looking for a pagination element.
:expectedresults: More than just the current page of errata can be selected
and applied, with changed per-page settings.
:customerscenario: true
:BZ: 1662254, 1846670
"""
org = module_sca_manifest_org
# custom_repo was created & added to cv, enabled in registered_contenthost.
repos = [
repo.read() for repo in module_cv.read().repository if repo.read().url == CUSTOM_REPO_3_URL
]
assert len(repos) > 0
custom_repo = repos[0]
custom_repo.sync()
# Set up and sync rh_repo
rh_repo_id = module_target_sat.api_factory.enable_sync_redhat_repo(
constants.REPOS['rhst8'],
module_sca_manifest_org.id,
)
# add rh_repo to cv, publish version and promote w/ the repository
module_target_sat.cli.ContentView.add_repository(
{
'id': module_cv.id,
'organization-id': org.id,
'repository-id': rh_repo_id,
}
)
module_cv = module_cv.read()
cv_publish_promote(
module_target_sat,
org,
module_cv,
module_lce,
)
registered_contenthost.add_rex_key(satellite=module_target_sat)
assert registered_contenthost.execute(r'subscription-manager repos --enable \*').status == 0
_chost_name = registered_contenthost.hostname
# Install all YUM 3 packages
pkgs = ' '.join(FAKE_3_YUM_OUTDATED_PACKAGES)
assert registered_contenthost.execute(f'yum install -y {pkgs}').status == 0
with session:
session.location.select(loc_name=DEFAULT_LOC)
# Go to new host's page UI, Content>Errata tab,
# There are two pagination objects on errata tab, we read the top one
pf4_pagination = session.host_new.get_errata_pagination(_chost_name)
assert pf4_pagination.read()
assert pf4_pagination.current_per_page == 20
# assert total_pages > 1 with default page settings
assert pf4_pagination.total_pages > 1
assert pf4_pagination.current_page == 1
assert pf4_pagination.total_items == registered_contenthost.applicable_errata_count
# Change per-page setting to 50, and assert there is now only one page
pf4_pagination.set_per_page(50)
pf4_pagination = session.host_new.get_errata_pagination(_chost_name)
assert pf4_pagination.read()
assert pf4_pagination.current_per_page == 50
assert pf4_pagination.current_page == 1
assert pf4_pagination.total_pages == 1
# assert at least the 28 errata from fake repo are present
assert pf4_pagination.total_items >= 28
_prior_app_count = pf4_pagination.total_items
# Change to a low per-page setting of 5
pf4_pagination.set_per_page(5)
pf4_pagination = session.host_new.get_errata_pagination(_chost_name)
assert pf4_pagination.read()
assert pf4_pagination.current_per_page == 5
assert pf4_pagination.current_page == 1
assert pf4_pagination.total_pages > 2
_prior_pagination = pf4_pagination.read()
# Install one available errata from UI with REX by default
errata_id = CUSTOM_REPO_3_ERRATA[1]
session.host_new.apply_erratas(_chost_name, f'errata_id="{errata_id}"')
# find host errata install job and status, timeout is 120s
# may take some time, wait for any not pending
status = module_target_sat.wait_for_tasks(
search_query=(f'Remote action: Install errata on {_chost_name} and result != pending'),
search_rate=2,
max_tries=60,
)
assert len(status) >= 1
task = status[0]
assert task.result == 'success'
assert 'host' in task.input
assert registered_contenthost.nailgun_host.id == task.input['host']['id']
# find bulk applicability task and status
status = module_target_sat.wait_for_tasks(
search_query=(
f'Bulk generate applicability for host {_chost_name} and result != pending'
),
search_rate=2,
max_tries=60,
)
assert len(status) >= 1
task = status[0]
assert task.result == 'success'
assert 'host_ids' in task.input
assert registered_contenthost.nailgun_host.id in task.input['host_ids']
# applicable errata is now one less
assert registered_contenthost.applicable_errata_count == _prior_app_count - 1
# wait for the tab to load with updated pagination, sat may be slow, timeout 30s.
# lambda: read is not the same as prior pagination read, and is also not empty {}.
_invalid_pagination = ({}, _prior_pagination)
session.browser.refresh()
wait_for(
lambda: (
session.host_new.get_errata_pagination(_chost_name).read()
not in _invalid_pagination
),
timeout=30,
delay=5,
)
# read updated pagination, handle slower UI loading
pf4_pagination = session.host_new.get_errata_pagination(_chost_name)
assert (_read_page := pf4_pagination.read())
assert _read_page != _prior_pagination
assert pf4_pagination.current_page == 1
# total_items decreased by one
item_count = pf4_pagination.total_items
assert item_count == _prior_app_count - 1
# Install All available from errata tab, we pass no search filter,
# so that all errata are selected, on all pages.
session.host_new.apply_erratas(_chost_name)
# find host's errata install job non-pending, timeout is 120s
status = module_target_sat.wait_for_tasks(
search_query=(f'Remote action: Install errata on {_chost_name} and result != pending'),
search_rate=2,
max_tries=60,
)
assert len(status) >= 1
task = status[0]
assert task.result == 'success'
assert 'host' in task.input
assert registered_contenthost.nailgun_host.id == task.input['host']['id']
# find bulk applicability task and status
status = module_target_sat.wait_for_tasks(
search_query=(
f'Bulk generate applicability for host {_chost_name} and result != pending'
),
search_rate=2,
max_tries=60,
)
assert len(status) >= 1
task = status[0]
assert task.result == 'success'
assert 'host_ids' in task.input
assert registered_contenthost.nailgun_host.id in task.input['host_ids']
# check there are no applicable errata left for Chost
assert registered_contenthost.applicable_errata_count == 0
# The errata table is not present when empty, it should not be paginated.
_items = -1
_ex_raised = False
session.browser.refresh()
try:
wait_for(
lambda: not session.host_new.get_errata_pagination(_chost_name).read(),
timeout=30,
delay=5,
)
except NoSuchElementException:
# pagination read raised exception, does not exist
_ex_raised = True
if not _ex_raised:
# pagination exists, reads empty {}, but we expect an
# exception when looking for a pagination element:
pf4_pagination = session.host_new.get_errata_pagination(_chost_name)
# exception trying to find element
with pytest.raises(NoSuchElementException):
_items = pf4_pagination.total_items
# assert nothing was found to update value
assert _items == -1, (
f'Found updated pagination total_items: {_items}, but expected to be empty.'
)
# would get failure at pytest.raises if no matching exception
_ex_raised = True
assert _ex_raised, (
'Search for empty pagination did not raise expected `NoSuchElementException`.'
)
@pytest.mark.skipif((not settings.robottelo.REPOS_HOSTING_URL), reason='Missing repos_hosting_url')
def test_positive_list(target_sat, session):
"""View all errata in an Org
:id: 71c7a054-a644-4c1e-b304-6bc34ea143f4
:setup:
1. two separate organizations, one custom product existing in each org.
:steps:
1. Create and sync separate repositories for each org.
2. Go to UI > Content Types > Errata page.
:expectedresults: Check that the errata belonging to one Org is not showing in the other.
:BZ: 1659941, 1837767
:customerscenario: true
"""
# new orgs, because module and function ones will overlap with other tests
org_0 = target_sat.api.Organization().create()
product_0 = target_sat.api.Product(organization=org_0).create()
org_1 = target_sat.api.Organization().create()
product_1 = target_sat.api.Product(organization=org_1).create()
# create and sync repository, for first org's errata
repo_0 = target_sat.api.Repository(
url=CUSTOM_REPO_URL,
product=product_0,
).create()
repo_0.sync()
# create and sync repo, for other org's errata
repo_1 = target_sat.api.Repository(
url=CUSTOM_REPO_3_URL,
product=product_1,
).create()
repo_1.sync()
with session:
# View in first organization
session.organization.select(org_name=org_0.name)
assert (
session.errata.search(CUSTOM_REPO_ERRATA_ID, applicable=False)[0]['Errata ID']
== CUSTOM_REPO_ERRATA_ID
), f'Could not find expected errata: {CUSTOM_REPO_ERRATA_ID}, in org: {org_0.name}.'
assert not session.errata.search(CUSTOM_REPO_3_ERRATA_ID, applicable=False), (
f'Found orgs ({org_1.name}) errata: {CUSTOM_REPO_3_ERRATA_ID},'
f' in other org ({org_0.name}) as well.'
)
# View in other organization
session.organization.select(org_name=org_1.name)
assert (
session.errata.search(CUSTOM_REPO_3_ERRATA_ID, applicable=False)[0]['Errata ID']
== CUSTOM_REPO_3_ERRATA_ID
), f'Could not find expected errata: {CUSTOM_REPO_3_ERRATA_ID}, in org: {org_1.name}.'
assert not session.errata.search(CUSTOM_REPO_ERRATA_ID, applicable=False), (
f'Found orgs ({org_0.name}) errata: {CUSTOM_REPO_ERRATA_ID},'
f' in other org ({org_1.name}) as well.'
)
def test_positive_list_permission(
test_name,
module_target_sat,
function_product,
function_sca_manifest_org,
):
"""Show errata only if the User has permissions to view them
:id: cdb28f6a-23df-47a2-88ab-cd3b492126b2
:Setup:
1. Create two products with one repo each. Sync them.
2. Make sure that they both have errata.
3. Create a user with view access on one product and not on the other.
:steps: Go to Content -> Errata.
:expectedresults: Check that the new user is able to see errata for one
product only.
"""
module_target_sat.api_factory.enable_sync_redhat_repo(
constants.REPOS['rhsclient9'],
function_sca_manifest_org.id,
)
custom_repo = module_target_sat.api.Repository(
url=CUSTOM_REPO_URL, product=function_product
).create()
custom_repo.sync()
# create role with access only to 'RHEL8' RedHat product
role = module_target_sat.api.Role(organization=[function_sca_manifest_org]).create()
module_target_sat.api.Filter(
permission=module_target_sat.api.Permission().search(
query={'search': 'resource_type="Katello::Product"'}
),
role=role,
search=f'name = "{PRDS["rhel9"]}"',
).create()
# generate login credentials for new role
user_password = gen_string('alphanumeric')
user = module_target_sat.api.User(
default_organization=function_sca_manifest_org,
organization=[function_sca_manifest_org],
role=[role],
password=user_password,
).create()
with module_target_sat.ui_session(
test_name, user=user.login, password=user_password
) as session:
# can view RHSC8 product's repo content (RHSC8 errata_id)
assert (
session.errata.search(REAL_RHSCLIENT_ERRATA, applicable=False)[0]['Errata ID']
== REAL_RHSCLIENT_ERRATA
)
# cannot view function product's custom repo content (fake custom errata_id)
assert not session.errata.search(CUSTOM_REPO_ERRATA_ID, applicable=False)
@pytest.mark.e2e
@pytest.mark.upgrade
@pytest.mark.no_containers
def test_positive_apply_for_all_hosts(
module_sca_manifest_org,
module_product,
target_sat,
module_lce,
module_cv,
module_ak,
session,
):
"""Apply an erratum for all content hosts
:id: d70a1bee-67f4-4883-a0b9-2ccc08a91738
:Setup: Errata synced on satellite server.
:customerscenario: true
:setup:
1. Create and sync one custom repo for all hosts, add to a content-view.
2. Checkout four contenthosts, latest rhel ver, via Broker.
3. Register all of the hosts to the same AK, CV, single repo.
:steps:
1. Go to Content -> Errata. Select an erratum -> Content Hosts tab.
2. Select all Content Hosts and apply the erratum.
:expectedresults:
1. Check invoked host tasks are successful.
2. Check that the erratum is applied in all the content hosts.
"""
num_hosts = 4
rhel_distro = target_sat.api_factory.supported_rhel_ver(
num=1,
prefix='rhel',
)
# one custom repo on satellite, for all hosts to use
custom_repo = target_sat.api.Repository(url=CUSTOM_REPO_URL, product=module_product).create()
custom_repo.sync()
module_cv.repository = [custom_repo]
module_cv.update(['repository'])
# Checkout hosts of the newest distro supported
with Broker(
nick=rhel_distro,
workflow='deploy-template',
host_class=ContentHost,
_count=num_hosts,
# TODO(@SatelliteQE/team-artemis): this is best effort for dualstack. This host deployment
# should be a part of a fixture
deploy_network_type=settings.content_host.network_type,
) as hosts:
if not isinstance(hosts, list) or len(hosts) != num_hosts:
pytest.fail('Failed to provision the expected number of hosts.')
for client in hosts:
# setup/register all hosts to same ak, content-view, and the one custom repo
setup = target_sat.api_factory.register_host_and_needed_setup(
organization=module_sca_manifest_org,
client=client,
activation_key=module_ak,
environment=module_lce,
content_view=module_cv,
enable_repos=True,
)
assert setup['result'] != 'error', f'{setup["message"]}'
assert (client := setup['client'])
assert client.subscribed
# install all outdated packages
pkgs = ' '.join(FAKE_9_YUM_OUTDATED_PACKAGES)
assert client.execute(f'yum install -y {pkgs}').status == 0
# update and check applicability
assert client.execute('subscription-manager repos').status == 0
assert client.applicable_errata_count > 0
assert client.applicable_package_count > 0
with session:
# possible in-progress applicability task(s), 60s margin
timestamp = (datetime.now(UTC).replace(microsecond=0) - timedelta(seconds=60)).strftime(
TIMESTAMP_FMT
)
session.location.select(loc_name=DEFAULT_LOC)
# for first errata, apply to all chosts at once,
# from ContentTypes > Errata > info > ContentHosts tab
errata_id = settings.repos.yum_9.errata[4] # RHBA-2012:1030
result = session.errata.install(
entity_name=errata_id,
host_names='All',
)
assert result['overall_status']['is_success']
# find single hosts job
# remote action tasks have lowercase errata_ids, ie: 'rhba-' not 'RHBA-'
hosts_job = target_sat.wait_for_tasks(
search_query=(
f'Run hosts job: Install errata {errata_id.lower()} and started_at >= "{timestamp}"'
),
search_rate=5,
max_tries=20,
)
assert len(hosts_job) == 1
# find multiple install tasks, one for each host
install_tasks = target_sat.wait_for_tasks(
search_query=(
f'Remote action: Install errata {errata_id.lower()} and started_at >= "{timestamp}"'
),
search_rate=5,
max_tries=20,
)
assert len(install_tasks) == num_hosts
# find bulk generate applicability task, and subtask for each host
applicability_tasks = target_sat.wait_for_tasks(
search_query=(
f'Bulk generate applicability for hosts and started_at >= "{timestamp}"'
),
search_rate=10,
max_tries=30,
)
assert len(applicability_tasks) > 0
# found updated kangaroo package in each host
updated_version = '0.2-1.noarch'
for client in hosts:
updated_pkg = session.host_new.get_packages(
entity_name=client.hostname, search='kangaroo'
)
assert len(updated_pkg) == 1
assert updated_pkg[0]['Installed version'] == updated_version
# for second errata, install in each chost and check, one at a time.
# from Host UI > details > Errata tab
for client in hosts:
# Navigate to All Hosts to ensure clean state before applying erratas
session.host_new.search(client.hostname)
session.host_new.apply_erratas(
entity_name=client.hostname,
search=f'errata_id=="{CUSTOM_REPO_ERRATA_ID}"',
)
# Wait for the errata installation task to complete
install_task = target_sat.wait_for_tasks(
search_query=(
f'Remote action: Install errata on {client.hostname} and result != pending'
),
search_rate=2,
max_tries=60,
)
assert len(install_task) >= 1
assert install_task[0].result == 'success'
# check updated package in chost details
assert client.execute('subscription-manager repos').status == 0
packages_rows = session.host_new.get_packages(
entity_name=client.hostname, search=FAKE_2_CUSTOM_PACKAGE
)
# updated walrus package found for each host
assert len(packages_rows) == 1
assert packages_rows[0]['Installed version'] == '5.21-1.noarch'
@pytest.mark.upgrade
@pytest.mark.rhel_ver_match('N-1')
def test_positive_view_cve(session, module_product, module_sca_manifest_org, target_sat):
"""View CVE number(s) in Errata Details page
:id: e1c2de13-fed8-448e-b618-c2adb6e82a35
:Setup: Errata synced on satellite server.
:steps: Go to Content -> Errata. Select an Errata.
:expectedresults:
1. Check if the CVE information is shown in Errata Details page.
2. Check if 'N/A' is displayed if CVE information is not present.