uncompressed oci image layers support added for make-rootfs action

Author: p5-vbnekit

src/_actions/make_rootfs/_oci_image/_fake.py

@@ -74,7 +74,7 @@ def _open_blob_reader(digest: str, source: _SourceReader, size: int):
         assert isinstance(size, int)
         _path = _parse_digest(value = digest).path
         with source.open(_path) as _stream:
-            with _make_size_guarantee(source = _stream, exact = size) as _guarantee: yield _guarantee
+            with _make_size_guarantee(source = _stream, limit = size) as _guarantee: yield _guarantee
 
     @contextlib.contextmanager
     def _open_blob_writer(image: str):
@@ -105,10 +105,11 @@ def _validate_layer_media_type(value: str):
         assert value
         assert value.startswith(_layer_media_type_begin)
         _value = value[_layer_media_type_minimum_size:]
-        if not _value: return value
-        assert "+" == _value[0]
-        _value = _value[1:]
-        assert _value in {"gzip", "bzip"}
+        if _value:
+            assert "+" == _value[0]
+            _value = _value[1:]
+            assert _value in {"gzip", "bzip"}
+        return value
 
     def _read_bundle(manifest: dict, source: _SourceReader):
         assert isinstance(manifest, dict)

src/_actions/make_rootfs/_oci_image/_size_guarantee.py

@@ -14,22 +14,34 @@ def left(self): return self.__left
 
         def tell(self, *args, **kwargs): return self.__stream.tell(*args, **kwargs)
 
+        def seek(self, *args, **kwargs):
+            try: return self.__stream.seek(*args, **kwargs)
+            finally: self.__finally()
+
         def read(self, *args, **kwargs):
-            _chunk = self.__stream.read(*args, **kwargs)
-            assert isinstance(_chunk, (str, bytes))
-            _size = len(_chunk)
-            if 0 < _size:
-                assert self.__left >= _size
-                self.__left -= _size
-            return _chunk
+            try: return self.__stream.read(*args, **kwargs)
+            finally: self.__finally()
 
         def __init__(self, stream, size: int):
             super().__init__()
             assert isinstance(size, int)
             assert 0 < size
+            _begin = stream.tell()
+            self.__end = size + _begin
             self.__left = size
+            self.__begin = _begin
             self.__stream = stream
 
+        def __finally(self):
+            _position = self.__stream.tell()
+
+            try:
+                assert self.__begin <= _position
+                assert self.__end >= _position
+            finally:
+                _position = max(self.__begin, min(self.__end, _position))
+                self.__left = min(self.__left, self.__end - _position)
+
     @contextlib.contextmanager
     def _data_wrapper(source: typing.Union[str, bytes], limit: typing.Optional[int], exact: typing.Optional[int]):
         assert isinstance(source, (str, bytes))
@@ -55,10 +67,13 @@ def _stream_wrapper(source, limit: typing.Optional[int], exact: typing.Optional[
 
         class _Class(object):
             @staticmethod
-            def read(*args, **kwargs): return _helper.read(*args, **kwargs)
+            def tell(*args, **kwargs): return _helper.tell(*args, **kwargs)
 
             @staticmethod
-            def tell(*args, **kwargs): return _helper.tell(*args, **kwargs)
+            def seek(*args, **kwargs): return _helper.seek(*args, **kwargs)
+
+            @staticmethod
+            def read(*args, **kwargs): return _helper.read(*args, **kwargs)
 
         yield _Class()
         assert (exact is None) or (0 == _helper.left)

src/_actions/make_rootfs/_oci_image/bundle/_class.py

@@ -100,7 +100,7 @@ def _open_layers(meta: typing.Iterable[dict], source: _SourceReader):
                 assert isinstance(_layer, dict)
                 _media = _layer["mediaType"]
                 assert isinstance(_media, str)
-                assert "application/vnd.oci.image.layer.v1.tar+gzip" == _media
+                assert _media in {"application/vnd.oci.image.layer.v1.tar", "application/vnd.oci.image.layer.v1.tar+gzip"}
                 _size = _layer["size"]
                 assert isinstance(_size, int)
                 assert 0 < _size