448
432
filename = filetype
449
433
size = randint(1024, 2048)
450
434
content = factory.make_string(size=size)
451
resource = factory.make_boot_resource(
435
resource = factory.make_BootResource(
452
436
rtype=BOOT_RESOURCE_TYPE.SYNCED, name=name,
453
437
architecture=architecture)
454
resource_set = factory.make_boot_resource_set(
438
resource_set = factory.make_BootResourceSet(
455
439
resource, version=version)
456
largefile = factory.make_large_file(content=content, size=size)
457
factory.make_boot_resource_file(
440
largefile = factory.make_LargeFile(content=content, size=size)
441
factory.make_BootResourceFile(
458
442
resource_set, largefile, filename=filename, filetype=filetype)
460
# Outside of the transaction, we run the actual test. The client will
461
# run inside of its own transaction, but once the streaming response
462
# is returned that transaction will be closed.
464
response = client.get(
466
'simplestreams_file_handler', kwargs={
472
'filename': filename,
475
# If TransactionWrapper does not work, then a ProgramError will be
476
# thrown. If it works then content will match.
477
self.assertEqual(content, b''.join(response.streaming_content))
478
self.assertTrue(largefile.content.closed)
443
return content, reverse(
444
'simplestreams_file_handler', kwargs={
450
'filename': filename,
453
def read_response(self, response):
454
"""Read the streaming_content from the response.
458
return b''.join(response.streaming_content)
460
def test_download_calls__get_new_connection(self):
461
content, url = self.make_file_for_client()
462
mock_get_new_connection = self.patch(
463
bootresources.ConnectionWrapper, '_get_new_connection')
466
response = client.get(url)
467
self.read_response(response)
468
self.assertThat(mock_get_new_connection, MockCalledOnceWith())
470
def test_download_connection_is_not_same_as_django_connections(self):
471
content, url = self.make_file_for_client()
473
class AssertConnectionWrapper(bootresources.ConnectionWrapper):
476
super(AssertConnectionWrapper, self)._set_up()
477
# Capture the created connection
478
AssertConnectionWrapper.connection = self._connection
481
# Close the stream, but we don't want to close the
482
# connection as the test is testing that the connection is
483
# not the same as the connection django is using for other
485
if self._stream is not None:
488
self._connection = None
491
bootresources, 'ConnectionWrapper', AssertConnectionWrapper)
494
response = client.get(url)
495
self.read_response(response)
497
# Add cleanup to close the connection, since this was removed from
498
# AssertConnectionWrapper.close method.
500
conn = AssertConnectionWrapper.connection
502
conn.leave_transaction_management()
504
self.addCleanup(close)
506
# The connection that is used by the wrapper cannot be the same as the
507
# connection be using for all other webrequests. Without this
508
# seperate the transactional middleware will fail to initialize,
509
# because the the connection will already be in a transaction.
511
# Note: cannot test if DatabaseWrapper != DatabaseWrapper, as it will
512
# report true, because the __eq__ operator only checks if the aliases
513
# are the same. This is checking the underlying connection is
514
# different, which is the important part.
516
connections["default"].connection,
517
AssertConnectionWrapper.connection.connection)
521
"""Make product dictionary that is just like the one provided
522
from simplsetreams."""
523
subarch = factory.make_name('subarch')
524
subarches = [factory.make_name('subarch') for _ in range(3)]
525
subarches.insert(0, subarch)
526
subarches = ','.join(subarches)
528
'os': factory.make_name('os'),
529
'arch': factory.make_name('arch'),
531
'release': factory.make_name('release'),
532
'kflavor': factory.make_name('kflavor'),
533
'subarches': subarches,
534
'version_name': factory.make_name('version'),
535
'label': factory.make_name('label'),
536
'ftype': factory.pick_enum(BOOT_RESOURCE_FILE_TYPE),
537
'kpackage': factory.make_name('kpackage'),
538
'di_version': factory.make_name('di_version'),
540
name = '%s/%s' % (product['os'], product['release'])
541
architecture = '%s/%s' % (product['arch'], product['subarch'])
542
return name, architecture, product
545
def make_boot_resource_group(
546
rtype=None, name=None, architecture=None,
547
version=None, filename=None, filetype=None):
548
"""Make boot resource that contains one set and one file."""
549
resource = factory.make_BootResource(
550
rtype=rtype, name=name, architecture=architecture)
551
resource_set = factory.make_BootResourceSet(resource, version=version)
552
rfile = factory.make_boot_resource_file_with_content(
553
resource_set, filename=filename, filetype=filetype)
554
return resource, resource_set, rfile
557
def make_boot_resource_group_from_product(product):
558
"""Make boot resource that contains one set and one file, using the
559
information from the given product.
561
The product dictionary is also updated to include the sha256 and size
562
for the created largefile. The calling function should use the returned
563
product in place of the passed product.
565
name = '%s/%s' % (product['os'], product['release'])
566
architecture = '%s/%s' % (product['arch'], product['subarch'])
567
resource = factory.make_BootResource(
568
rtype=BOOT_RESOURCE_TYPE.SYNCED, name=name,
569
architecture=architecture)
570
resource_set = factory.make_BootResourceSet(
571
resource, version=product['version_name'])
572
rfile = factory.make_boot_resource_file_with_content(
573
resource_set, filename=product['ftype'],
574
filetype=product['ftype'])
575
product['sha256'] = rfile.largefile.sha256
576
product['size'] = rfile.largefile.total_size
577
return product, resource
580
class TestBootResourceStore(MAASServerTestCase):
582
def make_boot_resources(self):
584
factory.make_BootResource(rtype=BOOT_RESOURCE_TYPE.SYNCED)
588
for resource in resources:
589
os, series = resource.name.split('/')
590
arch, subarch = resource.split_arch()
591
name = '%s/%s/%s/%s' % (os, arch, subarch, series)
592
resource_names.append(name)
593
return resources, resource_names
595
def test_init_initializes_variables(self):
596
_, resource_names = self.make_boot_resources()
597
store = BootResourceStore()
598
self.assertItemsEqual(resource_names, store._resources_to_delete)
599
self.assertEqual({}, store._content_to_finalize)
601
def test_prevent_resource_deletion_removes_resource(self):
602
resources, resource_names = self.make_boot_resources()
603
store = BootResourceStore()
604
resource = resources.pop()
606
store.prevent_resource_deletion(resource)
607
self.assertItemsEqual(resource_names, store._resources_to_delete)
609
def test_prevent_resource_deletion_doesnt_remove_unknown_resource(self):
610
resources, resource_names = self.make_boot_resources()
611
store = BootResourceStore()
612
resource = factory.make_BootResource(rtype=BOOT_RESOURCE_TYPE.SYNCED)
613
store.prevent_resource_deletion(resource)
614
self.assertItemsEqual(resource_names, store._resources_to_delete)
616
def test_save_content_later_adds_to__content_to_finalize_var(self):
617
_, _, rfile = make_boot_resource_group()
618
store = BootResourceStore()
619
store.save_content_later(rfile, sentinel.reader)
621
{rfile.id: sentinel.reader},
622
store._content_to_finalize)
624
def test_get_or_create_boot_resource_creates_resource(self):
625
name, architecture, product = make_product()
626
store = BootResourceStore()
627
resource = store.get_or_create_boot_resource(product)
628
self.assertEqual(BOOT_RESOURCE_TYPE.SYNCED, resource.rtype)
629
self.assertEqual(name, resource.name)
630
self.assertEqual(architecture, resource.architecture)
631
self.assertEqual(product['kflavor'], resource.extra['kflavor'])
632
self.assertEqual(product['subarches'], resource.extra['subarches'])
634
def test_get_or_create_boot_resource_gets_resource(self):
635
name, architecture, product = make_product()
636
expected = factory.make_BootResource(
637
rtype=BOOT_RESOURCE_TYPE.SYNCED, name=name,
638
architecture=architecture)
639
store = BootResourceStore()
640
resource = store.get_or_create_boot_resource(product)
641
self.assertEqual(expected, resource)
642
self.assertEqual(product['kflavor'], resource.extra['kflavor'])
643
self.assertEqual(product['subarches'], resource.extra['subarches'])
645
def test_get_or_create_boot_resource_calls_prevent_resource_deletion(self):
646
name, architecture, product = make_product()
647
resource = factory.make_BootResource(
648
rtype=BOOT_RESOURCE_TYPE.SYNCED,
649
name=name, architecture=architecture)
650
store = BootResourceStore()
651
mock_prevent = self.patch(store, 'prevent_resource_deletion')
652
store.get_or_create_boot_resource(product)
654
mock_prevent, MockCalledOnceWith(resource))
656
def test_get_or_create_boot_resource_converts_generated_into_synced(self):
657
name, architecture, product = make_product()
658
resource = factory.make_BootResource(
659
rtype=BOOT_RESOURCE_TYPE.GENERATED,
660
name=name, architecture=architecture)
661
store = BootResourceStore()
662
mock_prevent = self.patch(store, 'prevent_resource_deletion')
663
store.get_or_create_boot_resource(product)
665
BOOT_RESOURCE_TYPE.SYNCED,
666
reload_object(resource).rtype)
668
mock_prevent, MockNotCalled())
670
def test_get_or_create_boot_resource_set_creates_resource_set(self):
671
name, architecture, product = make_product()
672
product, resource = make_boot_resource_group_from_product(product)
673
resource.sets.all().delete()
674
store = BootResourceStore()
675
resource_set = store.get_or_create_boot_resource_set(resource, product)
676
self.assertEqual(product['version_name'], resource_set.version)
677
self.assertEqual(product['label'], resource_set.label)
679
def test_get_or_create_boot_resource_set_gets_resource_set(self):
680
name, architecture, product = make_product()
681
product, resource = make_boot_resource_group_from_product(product)
682
expected = resource.sets.first()
683
store = BootResourceStore()
684
resource_set = store.get_or_create_boot_resource_set(resource, product)
685
self.assertEqual(expected, resource_set)
686
self.assertEqual(product['label'], resource_set.label)
688
def test_get_or_create_boot_resource_file_creates_resource_file(self):
689
name, architecture, product = make_product()
690
product, resource = make_boot_resource_group_from_product(product)
691
resource_set = resource.sets.first()
692
resource_set.files.all().delete()
693
store = BootResourceStore()
694
rfile = store.get_or_create_boot_resource_file(resource_set, product)
695
self.assertEqual(product['ftype'], rfile.filename)
696
self.assertEqual(product['ftype'], rfile.filetype)
697
self.assertEqual(product['kpackage'], rfile.extra['kpackage'])
698
self.assertEqual(product['di_version'], rfile.extra['di_version'])
700
def test_get_or_create_boot_resource_file_gets_resource_file(self):
701
name, architecture, product = make_product()
702
product, resource = make_boot_resource_group_from_product(product)
703
resource_set = resource.sets.first()
704
expected = resource_set.files.first()
705
store = BootResourceStore()
706
rfile = store.get_or_create_boot_resource_file(resource_set, product)
707
self.assertEqual(expected, rfile)
708
self.assertEqual(product['ftype'], rfile.filetype)
709
self.assertEqual(product['kpackage'], rfile.extra['kpackage'])
710
self.assertEqual(product['di_version'], rfile.extra['di_version'])
712
def test_get_resource_file_log_identifier_returns_valid_ident(self):
713
os = factory.make_name('os')
714
series = factory.make_name('series')
715
arch = factory.make_name('arch')
716
subarch = factory.make_name('subarch')
717
version = factory.make_name('version')
718
filename = factory.make_name('filename')
719
name = '%s/%s' % (os, series)
720
architecture = '%s/%s' % (arch, subarch)
721
resource = factory.make_BootResource(
722
rtype=BOOT_RESOURCE_TYPE.SYNCED, name=name,
723
architecture=architecture)
724
resource_set = factory.make_BootResourceSet(
725
resource, version=version)
726
rfile = factory.make_boot_resource_file_with_content(
727
resource_set, filename=filename)
728
store = BootResourceStore()
730
'%s/%s/%s/%s/%s/%s' % (
731
os, arch, subarch, series, version, filename),
732
store.get_resource_file_log_identifier(rfile))
734
'%s/%s/%s/%s/%s/%s' % (
735
os, arch, subarch, series, version, filename),
736
store.get_resource_file_log_identifier(
737
rfile, resource_set, resource))
739
def test_write_content_saves_data(self):
740
rfile, reader, content = make_boot_resource_file_with_stream()
741
store = BootResourceStore()
742
store.write_content(rfile, reader)
743
self.assertTrue(BootResourceFile.objects.filter(id=rfile.id).exists())
744
with rfile.largefile.content.open('rb') as stream:
745
written_data = stream.read()
746
self.assertEqual(content, written_data)
748
def test_write_content_deletes_file_on_bad_checksum(self):
749
rfile, _, _ = make_boot_resource_file_with_stream()
750
reader = StringIO(factory.make_string())
751
store = BootResourceStore()
752
store.write_content(rfile, reader)
753
self.assertFalse(BootResourceFile.objects.filter(id=rfile.id).exists())
755
def test_finalize_calls_methods(self):
756
store = BootResourceStore()
757
mock_resource_cleaner = self.patch(store, 'resource_cleaner')
758
mock_perform_write = self.patch(store, 'perform_write')
759
mock_resource_set_cleaner = self.patch(store, 'resource_set_cleaner')
761
self.assertTrue(mock_resource_cleaner, MockCalledOnceWith())
762
self.assertTrue(mock_perform_write, MockCalledOnceWith())
763
self.assertTrue(mock_resource_set_cleaner, MockCalledOnceWith())
766
class TestBootResourceTransactional(TransactionTestCase):
767
"""Test methods on `BootResourceStore` that manage their own transactions.
769
This is done using TransactionTestCase so the database is flushed after
773
def test_insert_does_nothing_if_file_already_exists(self):
774
name, architecture, product = make_product()
775
with transaction.atomic():
776
product, resource = make_boot_resource_group_from_product(product)
777
rfile = resource.sets.first().files.first()
778
largefile = rfile.largefile
779
store = BootResourceStore()
780
mock_save_later = self.patch(store, 'save_content_later')
781
store.insert(product, sentinel.reader)
782
self.assertEqual(largefile, reload_object(rfile).largefile)
783
self.assertThat(mock_save_later, MockNotCalled())
785
def test_insert_uses_already_existing_largefile(self):
786
name, architecture, product = make_product()
787
with transaction.atomic():
788
product, resource = make_boot_resource_group_from_product(product)
789
resource_set = resource.sets.first()
790
resource_set.files.all().delete()
791
largefile = factory.make_LargeFile()
792
product['sha256'] = largefile.sha256
793
product['size'] = largefile.total_size
794
store = BootResourceStore()
795
mock_save_later = self.patch(store, 'save_content_later')
796
store.insert(product, sentinel.reader)
799
get_one(reload_object(resource_set).files.all()).largefile)
800
self.assertThat(mock_save_later, MockNotCalled())
802
def test_insert_deletes_mismatch_largefile(self):
803
name, architecture, product = make_product()
804
with transaction.atomic():
805
product, resource = make_boot_resource_group_from_product(product)
806
rfile = resource.sets.first().files.first()
807
delete_largefile = rfile.largefile
808
largefile = factory.make_LargeFile()
809
product['sha256'] = largefile.sha256
810
product['size'] = largefile.total_size
811
store = BootResourceStore()
812
mock_save_later = self.patch(store, 'save_content_later')
813
store.insert(product, sentinel.reader)
815
LargeFile.objects.filter(id=delete_largefile.id).exists())
816
self.assertEqual(largefile, reload_object(rfile).largefile)
817
self.assertThat(mock_save_later, MockNotCalled())
819
def test_insert_deletes_mismatch_largefile_keeps_other_resource_file(self):
820
name, architecture, product = make_product()
821
with transaction.atomic():
822
resource = factory.make_BootResource(
823
rtype=BOOT_RESOURCE_TYPE.SYNCED, name=name,
824
architecture=architecture)
825
resource_set = factory.make_BootResourceSet(
826
resource, version=product['version_name'])
827
other_type = factory.pick_enum(
828
BOOT_RESOURCE_FILE_TYPE, but_not=product['ftype'])
829
other_file = factory.make_boot_resource_file_with_content(
830
resource_set, filename=other_type, filetype=other_type)
831
rfile = factory.make_BootResourceFile(
832
resource_set, other_file.largefile,
833
filename=product['ftype'], filetype=product['ftype'])
834
largefile = factory.make_LargeFile()
835
product['sha256'] = largefile.sha256
836
product['size'] = largefile.total_size
837
store = BootResourceStore()
838
mock_save_later = self.patch(store, 'save_content_later')
839
store.insert(product, sentinel.reader)
840
self.assertEqual(largefile, reload_object(rfile).largefile)
842
LargeFile.objects.filter(id=other_file.largefile.id).exists())
844
BootResourceFile.objects.filter(id=other_file.id).exists())
846
other_file.largefile, reload_object(other_file).largefile)
847
self.assertThat(mock_save_later, MockNotCalled())
849
def test_insert_creates_new_largefile(self):
850
name, architecture, product = make_product()
851
with transaction.atomic():
852
resource = factory.make_BootResource(
853
rtype=BOOT_RESOURCE_TYPE.SYNCED, name=name,
854
architecture=architecture)
855
resource_set = factory.make_BootResourceSet(
856
resource, version=product['version_name'])
857
product['sha256'] = factory.make_string(size=64)
858
product['size'] = randint(1024, 2048)
859
store = BootResourceStore()
860
mock_save_later = self.patch(store, 'save_content_later')
861
store.insert(product, sentinel.reader)
862
rfile = get_one(reload_object(resource_set).files.all())
863
self.assertEqual(product['sha256'], rfile.largefile.sha256)
864
self.assertEqual(product['size'], rfile.largefile.total_size)
867
MockCalledOnceWith(rfile, sentinel.reader))
869
def test_resource_cleaner_removes_old_boot_resources(self):
870
with transaction.atomic():
872
factory.make_BootResource(rtype=BOOT_RESOURCE_TYPE.SYNCED)
875
store = BootResourceStore()
876
store.resource_cleaner()
877
for resource in resources:
878
os, series = resource.name.split('/')
879
arch, subarch = resource.split_arch()
881
BootResource.objects.has_synced_resource(
882
os, arch, subarch, series))
884
def test_resource_set_cleaner_removes_incomplete_set(self):
885
with transaction.atomic():
886
resource = factory.make_usable_boot_resource(
887
rtype=BOOT_RESOURCE_TYPE.SYNCED)
888
incomplete_set = factory.make_BootResourceSet(resource)
889
store = BootResourceStore()
890
store.resource_set_cleaner()
892
BootResourceSet.objects.filter(id=incomplete_set.id).exists())
894
def test_resource_set_cleaner_keeps_only_newest_completed_set(self):
895
with transaction.atomic():
896
resource = factory.make_BootResource(
897
rtype=BOOT_RESOURCE_TYPE.SYNCED)
898
old_complete_sets = []
900
resource_set = factory.make_BootResourceSet(resource)
901
factory.make_boot_resource_file_with_content(resource_set)
902
old_complete_sets.append(resource_set)
903
newest_set = factory.make_BootResourceSet(resource)
904
factory.make_boot_resource_file_with_content(newest_set)
905
store = BootResourceStore()
906
store.resource_set_cleaner()
907
self.assertItemsEqual([newest_set], resource.sets.all())
908
for resource_set in old_complete_sets:
910
BootResourceSet.objects.filter(id=resource_set.id).exists())
912
def test_resource_set_cleaner_removes_resources_with_empty_sets(self):
913
with transaction.atomic():
914
resource = factory.make_BootResource(
915
rtype=BOOT_RESOURCE_TYPE.SYNCED)
916
store = BootResourceStore()
917
store.resource_set_cleaner()
919
BootResource.objects.filter(id=resource.id).exists())
921
def test_perform_writes_writes_all_content(self):
922
with transaction.atomic():
923
files = [make_boot_resource_file_with_stream() for _ in range(3)]
924
store = BootResourceStore()
925
for rfile, reader, content in files:
926
store.save_content_later(rfile, reader)
927
store.perform_write()
928
with transaction.atomic():
929
for rfile, reader, content in files:
931
BootResourceFile.objects.filter(id=rfile.id).exists())
932
with rfile.largefile.content.open('rb') as stream:
933
written_data = stream.read()
934
self.assertEqual(content, written_data)
937
class TestImportImages(MAASTestCase):
939
def patch_and_capture_env_for_download_all_boot_resources(self):
941
"""Fake function; records a copy of the environment."""
943
def __call__(self, *args, **kwargs):
945
self.env = environ.copy()
947
capture = self.patch(
948
bootresources, 'download_all_boot_resources', CaptureEnv())
951
def test_download_boot_resources_syncs_repo(self):
952
fake_sync = self.patch(bootresources.BootResourceRepoWriter, 'sync')
953
store = BootResourceStore()
954
source_url = factory.make_url()
955
download_boot_resources(
956
source_url, store, None, None)
957
self.assertEqual(1, len(fake_sync.mock_calls))
959
def test_download_all_boot_resources_calls_download_boot_resources(self):
961
'url': factory.make_url(),
962
'keyring': self.make_file("keyring"),
964
product_mapping = ProductMapping()
965
store = BootResourceStore()
966
fake_download = self.patch(bootresources, 'download_boot_resources')
967
download_all_boot_resources(
968
sources=[source], product_mapping=product_mapping, store=store)
972
source['url'], store, product_mapping,
973
keyring_file=source['keyring']))
975
def test_download_all_boot_resources_calls_finalize_on_store(self):
976
product_mapping = ProductMapping()
977
store = BootResourceStore()
978
fake_finalize = self.patch(store, 'finalize')
979
download_all_boot_resources(
980
sources=[], product_mapping=product_mapping, store=store)
983
MockCalledOnceWith())
985
def test_has_synced_resources_returns_true(self):
986
factory.make_BootResource(rtype=BOOT_RESOURCE_TYPE.SYNCED)
987
self.assertTrue(bootresources.has_synced_resources())
989
def test_has_synced_resources_returns_false(self):
990
factory.make_BootResource(rtype=BOOT_RESOURCE_TYPE.UPLOADED)
991
self.assertFalse(bootresources.has_synced_resources())
993
def test__import_resources_exits_early_if_lock_held(self):
994
has_synced_resources = self.patch_autospec(
995
bootresources, "has_synced_resources")
996
with transaction.atomic():
997
with bootresources.locks.import_images:
998
bootresources._import_resources(force=True)
999
# The test for already-synced resources is not called if the
1000
# lock is already held.
1001
self.assertThat(has_synced_resources, MockNotCalled())
1003
def test__import_resources_exists_early_without_force(self):
1004
has_synced_resources = self.patch(
1005
bootresources, "has_synced_resources")
1006
bootresources._import_resources(force=False)
1007
# The test for already-synced resources is not performed if we're
1009
self.assertThat(has_synced_resources, MockCalledOnceWith())
1011
def test__import_resources_continues_with_force(self):
1012
has_synced_resources = self.patch(
1013
bootresources, "has_synced_resources")
1014
bootresources._import_resources(force=True)
1015
# The test for already-synced resources is performed if we're not
1017
self.assertThat(has_synced_resources, MockNotCalled())
1019
def test__import_resources_holds_lock(self):
1020
fake_write_all_keyrings = self.patch(
1021
bootresources, 'write_all_keyrings')
1023
def test_for_held_lock(directory, sources):
1024
self.assertTrue(bootresources.locks.import_images.is_locked())
1026
fake_write_all_keyrings.side_effect = test_for_held_lock
1028
bootresources._import_resources(force=True)
1029
self.assertFalse(bootresources.locks.import_images.is_locked())
1031
def test__import_resources_calls_functions_with_correct_parameters(self):
1032
fake_write_all_keyrings = self.patch(
1033
bootresources, 'write_all_keyrings')
1034
fake_write_all_keyrings.return_value = sentinel.sources
1035
fake_image_descriptions = self.patch(
1036
bootresources, 'download_all_image_descriptions')
1037
descriptions = Mock()
1038
descriptions.is_empty.return_value = False
1039
fake_image_descriptions.return_value = descriptions
1040
fake_map_products = self.patch(
1041
bootresources, 'map_products')
1042
fake_map_products.return_value = sentinel.mapping
1043
fake_download_all_boot_resources = self.patch(
1044
bootresources, 'download_all_boot_resources')
1046
bootresources._import_resources(force=True)
1049
fake_write_all_keyrings,
1050
MockCalledOnceWith(ANY, []))
1052
fake_image_descriptions,
1053
MockCalledOnceWith(sentinel.sources))
1056
MockCalledOnceWith(descriptions))
1058
fake_download_all_boot_resources,
1059
MockCalledOnceWith(sentinel.sources, sentinel.mapping))
1061
def test__import_resources_has_env_GNUPGHOME_set(self):
1062
fake_image_descriptions = self.patch(
1063
bootresources, 'download_all_image_descriptions')
1064
descriptions = Mock()
1065
descriptions.is_empty.return_value = False
1066
fake_image_descriptions.return_value = descriptions
1067
self.patch(bootresources, 'map_products')
1068
capture = self.patch_and_capture_env_for_download_all_boot_resources()
1070
bootresources._import_resources(force=True)
1072
bootresources.get_maas_user_gpghome(),
1073
capture.env['GNUPGHOME'])
1075
def test__import_resources_has_env_http_and_https_proxy_set(self):
1076
proxy_address = factory.make_name('proxy')
1077
Config.objects.set_config('http_proxy', proxy_address)
1079
fake_image_descriptions = self.patch(
1080
bootresources, 'download_all_image_descriptions')
1081
descriptions = Mock()
1082
descriptions.is_empty.return_value = False
1083
fake_image_descriptions.return_value = descriptions
1084
self.patch(bootresources, 'map_products')
1085
capture = self.patch_and_capture_env_for_download_all_boot_resources()
1087
bootresources._import_resources(force=True)
1089
(proxy_address, proxy_address),
1090
(capture.env['http_proxy'], capture.env['http_proxy']))
1092
def test__import_resources_calls_import_boot_images_on_clusters(self):
1093
nodegroup = MagicMock()
1094
self.patch(bootresources, 'NodeGroup', nodegroup)
1096
fake_image_descriptions = self.patch(
1097
bootresources, 'download_all_image_descriptions')
1098
descriptions = Mock()
1099
descriptions.is_empty.return_value = False
1100
fake_image_descriptions.return_value = descriptions
1101
self.patch(bootresources, 'map_products')
1102
self.patch(bootresources, 'download_all_boot_resources')
1104
bootresources._import_resources(force=True)
1106
nodegroup.objects.import_boot_images_on_accepted_clusters,
1107
MockCalledOnceWith())
added
removed
src/maasserver/tests/test_bootresources.py
