rules/scripts/bitstreams_workspace_test.py - 3p/lowrisc/opentitan - Git at Google

 # Copyright lowRISC contributors.
 # Licensed under the Apache License, Version 2.0, see LICENSE for details.
 # SPDX-License-Identifier: Apache-2.0

 import os
 import unittest
 import unittest.mock

 from bitstreams_workspace import BitstreamCache


 class TestBitstreamCache(unittest.TestCase):

     def test_make_with_default(self):
         # Changes to command-line argument defaults could break this method, so
         # it's important to at least have code coverage.
         BitstreamCache.MakeWithDefaults()

     def test_get_from_cache(self):
         BITSTREAM_ORIG = 'lowrisc_systems_chip_earlgrey_cw310_0.1.bit.orig'
         BITSTREAM_SPLICE = 'lowrisc_systems_chip_earlgrey_cw310_0.1.bit.splice'

         MOCKED_OS_WALK_RETURN = [
             # os.walk() yields tuples of the form (root, dir, files).
             ('cache/abcd', [],
              [BITSTREAM_ORIG, BITSTREAM_SPLICE, 'rom.mmi', 'otp.mmi']),
         ]
         os.walk = unittest.mock.MagicMock(name='os.walk',
                                           return_value=MOCKED_OS_WALK_RETURN)

         cache = BitstreamCache('/',
                                '/tmp/cache/opentitan-bitstreams',
                                'latest.txt',
                                offline=True)
         cache.InitRepository = unittest.mock.MagicMock(name='method')

         cached_files = cache.GetFromCache('abcd')

         # This is more of an implementation detail, but it verifies that we hit
         # the the mocked `os.walk` function as expected.
         os.walk.assert_called_once_with('cache/abcd')

         self.assertEqual(
             dict(cached_files), {
                 "schemaVersion": 1,
                 "buildId": "abcd",
                 "outputFiles": {
                     BITSTREAM_ORIG: {
                         "buildTarget": "//hw/bitstream/vivado:fpga_cw310",
                         "outputInfo": {
                             "@type": "bitstreamInfo",
                             "design": "chip_earlgrey_cw310"
                         }
                     },
                     "rom.mmi": {
                         "buildTarget": "//hw/bitstream/vivado:fpga_cw310",
                         "outputInfo": {
                             "@type": "memoryMapInfo",
                             "design": "chip_earlgrey_cw310",
                             "memoryId": "rom",
                         }
                     },
                     "otp.mmi": {
                         "buildTarget": "//hw/bitstream/vivado:fpga_cw310",
                         "outputInfo": {
                             "@type": "memoryMapInfo",
                             "design": "chip_earlgrey_cw310",
                             "memoryId": "otp",
                         }
                     },
                 },
             })

         os.walk.assert_called_once_with('cache/abcd')

     def test_write_build_file(self):
         BITSTREAM_ORIG = 'lowrisc_systems_chip_earlgrey_cw310_0.1.bit.orig'
         BITSTREAM_SPLICE = 'lowrisc_systems_chip_earlgrey_cw310_0.1.bit.splice'

         MOCKED_OS_WALK_RETURN = [
             # os.walk() yields tuples of the form (root, dir, files).
             ('cache/abcd', [],
              [BITSTREAM_ORIG, BITSTREAM_SPLICE, 'rom.mmi', 'otp.mmi']),
         ]
         os.walk = unittest.mock.MagicMock(name='os.walk',
                                           return_value=MOCKED_OS_WALK_RETURN)

         BitstreamCache._GetDateTimeStr = unittest.mock.MagicMock(
             name='BitstreamCache._GetDateTimeStr',
             return_value='2022-07-14T15:02:54.463801')

         cache = BitstreamCache('/',
                                '/tmp/cache/opentitan-bitstreams',
                                'latest.txt',
                                offline=True)
         cache.InitRepository = unittest.mock.MagicMock(name='method')

         bazel_string = cache._ConstructBazelString('BUILD.mock', 'abcd')
         self.maxDiff = None
         self.assertEqual(
             bazel_string, '''# This file was autogenerated. Do not edit!
 # Built at 2022-07-14T15:02:54.463801.
 # Configured for bitstream: abcd

 package(default_visibility = ["//visibility:public"])

 exports_files(glob(["cache/**"]))

 filegroup(
     name = "chip_earlgrey_cw310_bitstream",
     srcs = ["cache/abcd/lowrisc_systems_chip_earlgrey_cw310_0.1.bit.orig"],
 )

 filegroup(
     name = "chip_earlgrey_cw310_otp_mmi",
     srcs = ["cache/abcd/otp.mmi"],
 )

 filegroup(
     name = "chip_earlgrey_cw310_rom_mmi",
     srcs = ["cache/abcd/rom.mmi"],
 )
 ''')

         # This is more of an implementation detail, but it verifies that we hit
         # the the mocked `os.walk` function as expected.
         os.walk.assert_called_once_with('cache/abcd')


 class TestFetchAvailableBitstreams(unittest.TestCase):
     """
     The BitstreamCache.GetBitstreamsAvailable method calls the XML API on
     the root level of the GCP bucket to get the list of files in the
     bitstream cache. If this list is sufficiently long, the response will
     be paginated, as indicated by the <IsTruncated> tag. For p pages,
     GetBitstreamsAvailable is expected to make p+1 calls to the API (using
     BitstreamCache.Get). The last call is to get latest.txt which indicates
     the latest bitstream.

     The XML response elements are documented here:
     https://cloud.google.com/storage/docs/xml-api/get-bucket-list#response_body_elements
     """

     def setUp(self):
         self.cache = BitstreamCache('/', '/tmp/cache/opentitan-bitstreams',
                                     'latest.txt')
         self.cache.InitRepository = unittest.mock.MagicMock(name='method')

     def test_fetch_available_bitstreams_single_page(self):
         """Test fetching the available bitstreams without pagination."""

         MOCKED_GET_RETURN = [
             b"""<ListBucketResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
 <Name>opentitan-bitstreams</Name>
 <Prefix/>
 <Marker/>
 <IsTruncated>false</IsTruncated>
 <Contents>
 <Key>master/bitstream-0.tar.gz</Key>
 <Generation>1669083850593267</Generation>
 <MetaGeneration>1</MetaGeneration>
 <LastModified>2022-11-22T02:24:10.633Z</LastModified>
 <ETag>"e82b93a0f88161e594ef79f41277de92"</ETag>
 <Size>18845320</Size>
 </Contents>
 </ListBucketResult>""", b"""2022-12-01T14:54:13
 0"""
         ]

         # Mock out the Get method to simulate a network access
         self.cache.Get = unittest.mock.MagicMock(
             name='cache.Get',
             side_effect=MOCKED_GET_RETURN,
         )
         self.cache.GetBitstreamsAvailable(refresh=True)
         self.assertEqual(self.cache.Get.call_count, 2)
         self.assertEqual(self.cache.available, {
             "0": "master/bitstream-0.tar.gz",
             "latest": "0",
         })

     def test_fetch_available_bitstreams_with_pagination(self):
         """Test fetching the XML file with the list of available bitstreams."""
         MOCKED_GET_RETURN = [
             b"""<ListBucketResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
 <Name>opentitan-bitstreams</Name>
 <Prefix/>
 <Marker/>
 <NextMarker>master/bitstream-1.tar.gz</NextMarker>
 <IsTruncated>true</IsTruncated>
 <Contents>
 <Key>master/bitstream-0.tar.gz</Key>
 <Generation>1656382040594268</Generation>
 <MetaGeneration>1</MetaGeneration>
 <LastModified>2022-06-28T02:07:20.635Z</LastModified>
 <ETag>"5b6f3f9ef43f67b988cac31c73949e85"</ETag>
 <Size>12254300</Size>
 </Contents>
 </ListBucketResult>""",
             b"""<ListBucketResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
 <Name>opentitan-bitstreams</Name>
 <Prefix/>
 <Marker>master/bitstream-1.tar.gz</Marker>
 <IsTruncated>false</IsTruncated>
 <Contents>
 <Key>master/bitstream-1.tar.gz</Key>
 <Generation>1656382040594268</Generation>
 <MetaGeneration>1</MetaGeneration>
 <LastModified>2022-06-28T02:07:20.635Z</LastModified>
 <ETag>"5b6f3f9ef43f67b988cac31c73949e85"</ETag>
 <Size>12254300</Size>
 </Contents>
 <Contents>
 <Key>master/latest.txt</Key>
 <Generation>1669836798495359</Generation>
 <MetaGeneration>1</MetaGeneration>
 <LastModified>2022-11-30T19:33:18.615Z</LastModified>
 <ETag>"58498757ff6f02bcbfbae67eb92dfa4b"</ETag>
 <Size>60</Size>
 </Contents>
 </ListBucketResult>
 """, b"""2022-12-01T14:54:13
 1"""
         ]

         # Mock out the Get method to simulate a network access
         self.cache.Get = unittest.mock.MagicMock(
             name='cache.Get',
             side_effect=MOCKED_GET_RETURN,
         )
         self.cache.GetBitstreamsAvailable(refresh=True)
         self.assertEqual(self.cache.Get.call_count, 3)
         self.assertEqual(
             self.cache.available, {
                 "0": "master/bitstream-0.tar.gz",
                 "1": "master/bitstream-1.tar.gz",
                 "latest": "1",
             })


 if __name__ == '__main__':
     unittest.main()
	# Copyright lowRISC contributors.
	# Licensed under the Apache License, Version 2.0, see LICENSE for details.
	# SPDX-License-Identifier: Apache-2.0

	import os
	import unittest
	import unittest.mock

	from bitstreams_workspace import BitstreamCache


	class TestBitstreamCache(unittest.TestCase):

	def test_make_with_default(self):
	# Changes to command-line argument defaults could break this method, so
	# it's important to at least have code coverage.
	BitstreamCache.MakeWithDefaults()

	def test_get_from_cache(self):
	BITSTREAM_ORIG = 'lowrisc_systems_chip_earlgrey_cw310_0.1.bit.orig'
	BITSTREAM_SPLICE = 'lowrisc_systems_chip_earlgrey_cw310_0.1.bit.splice'

	MOCKED_OS_WALK_RETURN = [
	# os.walk() yields tuples of the form (root, dir, files).
	('cache/abcd', [],
	[BITSTREAM_ORIG, BITSTREAM_SPLICE, 'rom.mmi', 'otp.mmi']),
	]
	os.walk = unittest.mock.MagicMock(name='os.walk',
	return_value=MOCKED_OS_WALK_RETURN)

	cache = BitstreamCache('/',
	'/tmp/cache/opentitan-bitstreams',
	'latest.txt',
	offline=True)
	cache.InitRepository = unittest.mock.MagicMock(name='method')

	cached_files = cache.GetFromCache('abcd')

	# This is more of an implementation detail, but it verifies that we hit
	# the the mocked `os.walk` function as expected.
	os.walk.assert_called_once_with('cache/abcd')

	self.assertEqual(
	dict(cached_files), {
	"schemaVersion": 1,
	"buildId": "abcd",
	"outputFiles": {
	BITSTREAM_ORIG: {
	"buildTarget": "//hw/bitstream/vivado:fpga_cw310",
	"outputInfo": {
	"@type": "bitstreamInfo",
	"design": "chip_earlgrey_cw310"
	}
	},
	"rom.mmi": {
	"buildTarget": "//hw/bitstream/vivado:fpga_cw310",
	"outputInfo": {
	"@type": "memoryMapInfo",
	"design": "chip_earlgrey_cw310",
	"memoryId": "rom",
	}
	},
	"otp.mmi": {
	"buildTarget": "//hw/bitstream/vivado:fpga_cw310",
	"outputInfo": {
	"@type": "memoryMapInfo",
	"design": "chip_earlgrey_cw310",
	"memoryId": "otp",
	}
	},
	},
	})

	os.walk.assert_called_once_with('cache/abcd')

	def test_write_build_file(self):
	BITSTREAM_ORIG = 'lowrisc_systems_chip_earlgrey_cw310_0.1.bit.orig'
	BITSTREAM_SPLICE = 'lowrisc_systems_chip_earlgrey_cw310_0.1.bit.splice'

	MOCKED_OS_WALK_RETURN = [
	# os.walk() yields tuples of the form (root, dir, files).
	('cache/abcd', [],
	[BITSTREAM_ORIG, BITSTREAM_SPLICE, 'rom.mmi', 'otp.mmi']),
	]
	os.walk = unittest.mock.MagicMock(name='os.walk',
	return_value=MOCKED_OS_WALK_RETURN)

	BitstreamCache._GetDateTimeStr = unittest.mock.MagicMock(
	name='BitstreamCache._GetDateTimeStr',
	return_value='2022-07-14T15:02:54.463801')

	cache = BitstreamCache('/',
	'/tmp/cache/opentitan-bitstreams',
	'latest.txt',
	offline=True)
	cache.InitRepository = unittest.mock.MagicMock(name='method')

	bazel_string = cache._ConstructBazelString('BUILD.mock', 'abcd')
	self.maxDiff = None
	self.assertEqual(
	bazel_string, '''# This file was autogenerated. Do not edit!
	# Built at 2022-07-14T15:02:54.463801.
	# Configured for bitstream: abcd

	package(default_visibility = ["//visibility:public"])

	exports_files(glob(["cache/**"]))

	filegroup(
	name = "chip_earlgrey_cw310_bitstream",
	srcs = ["cache/abcd/lowrisc_systems_chip_earlgrey_cw310_0.1.bit.orig"],
	)

	filegroup(
	name = "chip_earlgrey_cw310_otp_mmi",
	srcs = ["cache/abcd/otp.mmi"],
	)

	filegroup(
	name = "chip_earlgrey_cw310_rom_mmi",
	srcs = ["cache/abcd/rom.mmi"],
	)
	''')

	# This is more of an implementation detail, but it verifies that we hit
	# the the mocked `os.walk` function as expected.
	os.walk.assert_called_once_with('cache/abcd')


	class TestFetchAvailableBitstreams(unittest.TestCase):
	"""
	The BitstreamCache.GetBitstreamsAvailable method calls the XML API on
	the root level of the GCP bucket to get the list of files in the
	bitstream cache. If this list is sufficiently long, the response will
	be paginated, as indicated by the <IsTruncated> tag. For p pages,
	GetBitstreamsAvailable is expected to make p+1 calls to the API (using
	BitstreamCache.Get). The last call is to get latest.txt which indicates
	the latest bitstream.

	The XML response elements are documented here:
	https://cloud.google.com/storage/docs/xml-api/get-bucket-list#response_body_elements
	"""

	def setUp(self):
	self.cache = BitstreamCache('/', '/tmp/cache/opentitan-bitstreams',
	'latest.txt')
	self.cache.InitRepository = unittest.mock.MagicMock(name='method')

	def test_fetch_available_bitstreams_single_page(self):
	"""Test fetching the available bitstreams without pagination."""

	MOCKED_GET_RETURN = [
	b"""<ListBucketResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
	<Name>opentitan-bitstreams</Name>
	<Prefix/>
	<Marker/>
	<IsTruncated>false</IsTruncated>
	<Contents>
	<Key>master/bitstream-0.tar.gz</Key>
	<Generation>1669083850593267</Generation>
	<MetaGeneration>1</MetaGeneration>
	<LastModified>2022-11-22T02:24:10.633Z</LastModified>
	<ETag>"e82b93a0f88161e594ef79f41277de92"</ETag>
	<Size>18845320</Size>
	</Contents>
	</ListBucketResult>""", b"""2022-12-01T14:54:13
	0"""
	]

	# Mock out the Get method to simulate a network access
	self.cache.Get = unittest.mock.MagicMock(
	name='cache.Get',
	side_effect=MOCKED_GET_RETURN,
	)
	self.cache.GetBitstreamsAvailable(refresh=True)
	self.assertEqual(self.cache.Get.call_count, 2)
	self.assertEqual(self.cache.available, {
	"0": "master/bitstream-0.tar.gz",
	"latest": "0",
	})

	def test_fetch_available_bitstreams_with_pagination(self):
	"""Test fetching the XML file with the list of available bitstreams."""
	MOCKED_GET_RETURN = [
	b"""<ListBucketResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
	<Name>opentitan-bitstreams</Name>
	<Prefix/>
	<Marker/>
	<NextMarker>master/bitstream-1.tar.gz</NextMarker>
	<IsTruncated>true</IsTruncated>
	<Contents>
	<Key>master/bitstream-0.tar.gz</Key>
	<Generation>1656382040594268</Generation>
	<MetaGeneration>1</MetaGeneration>
	<LastModified>2022-06-28T02:07:20.635Z</LastModified>
	<ETag>"5b6f3f9ef43f67b988cac31c73949e85"</ETag>
	<Size>12254300</Size>
	</Contents>
	</ListBucketResult>""",
	b"""<ListBucketResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
	<Name>opentitan-bitstreams</Name>
	<Prefix/>
	<Marker>master/bitstream-1.tar.gz</Marker>
	<IsTruncated>false</IsTruncated>
	<Contents>
	<Key>master/bitstream-1.tar.gz</Key>
	<Generation>1656382040594268</Generation>
	<MetaGeneration>1</MetaGeneration>
	<LastModified>2022-06-28T02:07:20.635Z</LastModified>
	<ETag>"5b6f3f9ef43f67b988cac31c73949e85"</ETag>
	<Size>12254300</Size>
	</Contents>
	<Contents>
	<Key>master/latest.txt</Key>
	<Generation>1669836798495359</Generation>
	<MetaGeneration>1</MetaGeneration>
	<LastModified>2022-11-30T19:33:18.615Z</LastModified>
	<ETag>"58498757ff6f02bcbfbae67eb92dfa4b"</ETag>
	<Size>60</Size>
	</Contents>
	</ListBucketResult>
	""", b"""2022-12-01T14:54:13
	1"""
	]

	# Mock out the Get method to simulate a network access
	self.cache.Get = unittest.mock.MagicMock(
	name='cache.Get',
	side_effect=MOCKED_GET_RETURN,
	)
	self.cache.GetBitstreamsAvailable(refresh=True)
	self.assertEqual(self.cache.Get.call_count, 3)
	self.assertEqual(
	self.cache.available, {
	"0": "master/bitstream-0.tar.gz",
	"1": "master/bitstream-1.tar.gz",
	"latest": "1",
	})


	if __name__ == '__main__':
	unittest.main()