File: test_stream_nwstorage.py

package info (click to toggle)
py7zr 1.0.0%2Bdfsg1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 6,352 kB
  • sloc: python: 9,009; sh: 360; makefile: 197; ansic: 35
file content (69 lines) | stat: -rw-r--r-- 2,364 bytes parent folder | download
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
import os
import pathlib
from typing import Optional

import requests
from pytest_httpserver import HTTPServer

import py7zr
import py7zr.io

testdata_path = pathlib.Path(os.path.dirname(__file__)).joinpath("data")
os.umask(0o022)


def test_extract_stream(httpserver: HTTPServer):
    httpserver.expect_request("/setup.cfg", method="PUT").respond_with_data("ok")
    httpserver.expect_request("/setup.cfg", method="GET").respond_with_data(
        b"[flake8]\nmax-line-length = 125\n\n[bdist_wheel]\nuniversal=1\n"
    )
    httpserver.expect_request("/setup.py", method="PUT").respond_with_data("ok")
    httpserver.expect_request("/scripts/py7zr", method="PUT").respond_with_data("ok")
    factory = StreamWriterFactory(httpserver)
    with py7zr.SevenZipFile(testdata_path.joinpath("test_1.7z").open(mode="rb")) as archive:
        archive.extractall(factory=factory)
    assert len(factory.products) == 3
    assert "setup.cfg" in factory.products.keys()
    assert "setup.py" in factory.products.keys()
    assert "scripts/py7zr" in factory.products.keys()
    assert factory.products["setup.cfg"].read() == b"[flake8]\nmax-line-length = 125\n\n[bdist_wheel]\nuniversal=1\n"


class StreamWriter(py7zr.io.Py7zIO):
    """Pseudo object storage writer."""

    def __init__(self, httpserver: HTTPServer, fname: str):
        self.httpserver: HTTPServer = httpserver
        self.fname = fname
        self.length = 0

    def write(self, data: [bytes, bytearray]):
        self.length += len(data)
        requests.put(self.httpserver.url_for(self.fname), data=data)
        self.httpserver.check_assertions()

    def read(self, size: Optional[int] = None) -> bytes:
        response = requests.get(self.httpserver.url_for(self.fname))
        return response.content

    def seek(self, offset: int, whence: int = 0) -> int:
        return offset

    def flush(self) -> None:
        pass

    def size(self) -> int:
        return self.length


class StreamWriterFactory(py7zr.io.WriterFactory):
    """Factory class to return StreamWriter object."""

    def __init__(self, httpserver: HTTPServer):
        self.httpserver: HTTPServer = httpserver
        self.products = {}

    def create(self, filename: str) -> py7zr.io.Py7zIO:
        product = StreamWriter(self.httpserver, filename)
        self.products[filename] = product
        return product