Skip to content

mock

MockDrsClient

Bases: DrsClient

Simulate responses from server.

Source code in drs_downloader/clients/mock.py 
 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
class MockDrsClient(DrsClient):
    """Simulate responses from server."""

    async def sign_url(self, drs_object: DrsObject, verbose: bool = False) -> Optional[DrsObject]:
        """Simulate url signing by waiting 1-3 seconds, return populated DrsObject

        Args:
            drs_object:

        Returns:
            populated DrsObject
        """
        # simulate a failed signature
        if drs_object.id == BAD_SIGNATURE:
            return None

        # here we sleep while the file is open and measure total files open
        fp = tempfile.TemporaryFile()
        sleep_duration = random.randint(1, 3)
        await asyncio.sleep(delay=sleep_duration)
        fp.write(b"sign url")
        self.statistics.set_max_files_open()
        fp.close()

        # provide expected result, e.g. X-Signature
        access_url = f"{drs_object.self_uri}?X-Signature={uuid.uuid1()}"
        # place it in the right spot in the drs object
        drs_object.access_methods.append(AccessMethod(access_url=access_url, type="gs"))

        return drs_object

    async def download_part(
        self, drs_object: DrsObject, start: int, size: int, destination_path: Path, verbose: bool = False
    ) -> Path:
        """Actually download a part.

        Args:
            destination_path:
            drs_object:
            start:
            size:

        Returns:
            full path to that part.
        """

        # calculate actual part size from range see https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Range

        length_ = size - start + 1
        # logger.info((drs_object.name, start, length_))
        # logger.error("ERROR1 %s",drs_object)

        if BAD_ID in drs_object.self_uri:
            logger.warning(f"Mock bad id {drs_object.self_uri}")
            drs_object.errors = ["Mock error BAD_ID"]
            return None

        with open(Path(os.getcwd(), f"{drs_object.name}.golden"), "rb") as f:
            f.seek(start)
            data = f.read(length_)

        (fd, name,) = tempfile.mkstemp(
            prefix=f"{drs_object.name}.{start}.{size}.",
            suffix=".part",
            dir=str(destination_path),
        )
        with os.fdopen(fd, "wb") as fp:
            sleep_duration = random.randint(1, 3)
            await asyncio.sleep(delay=sleep_duration)
            fp.write(data)
            self.statistics.set_max_files_open()
            fp.close()

        return Path(name)

    async def get_object(self, object_id: str, verbose: bool = False) -> DrsObject:
        """Fetch the object from repository DRS Service.

        See https://ga4gh.github.io/data-repository-service-schemas/preview/release/drs-1.0.0/docs/#_getobject.

        Args:
            object_id:

        Returns:

        """
        # """Actually fetch the object.
        #
        # """
        fp = tempfile.TemporaryFile()
        sleep_duration = random.randint(1, 3)
        await asyncio.sleep(delay=sleep_duration)
        fp.write(b"get object")
        self.statistics.set_max_files_open()
        fp.close()

        id_ = str(uuid.uuid4())
        name_ = f"file-{id_}.txt"

        line = b"Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n"  # noqa
        line_len = len(line)
        number_of_lines = int(random.randint(line_len, MAX_SIZE_OF_OBJECT) / line_len)
        lines = line * number_of_lines
        size_ = len(lines)

        # write it for testing
        destination_dir = Path(os.getcwd())
        destination_dir.mkdir(parents=True, exist_ok=True)
        with open(Path(f"{destination_dir}/{name_}.golden"), "wb") as f:
            f.write(lines)

        checksum = Checksum(hashlib.new("md5", lines).hexdigest(), type="md5")

        # simulate an incorrect MD5
        if object_id == BAD_MD5:
            checksum = Checksum(hashlib.new("md5", line).hexdigest(), type="md5")

        # simulate an incorrect size
        if object_id == INCORRECT_SIZE:
            size_ += 1000

        return DrsObject(
            self_uri=object_id,
            size=size_,
            # md5, etag, crc32c, trunc512, or sha1
            checksums=[checksum],
            id=id_,
            name=name_,
        )

download_part(drs_object, start, size, destination_path, verbose=False) async

Actually download a part.

Parameters:

Name Type Description Default
destination_path Path
required
drs_object DrsObject
required
start int
required
size int
required

Returns:

Type Description
Path

full path to that part.
Source code in drs_downloader/clients/mock.py 
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
async def download_part(
    self, drs_object: DrsObject, start: int, size: int, destination_path: Path, verbose: bool = False
) -> Path:
    """Actually download a part.

    Args:
        destination_path:
        drs_object:
        start:
        size:

    Returns:
        full path to that part.
    """

    # calculate actual part size from range see https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Range

    length_ = size - start + 1
    # logger.info((drs_object.name, start, length_))
    # logger.error("ERROR1 %s",drs_object)

    if BAD_ID in drs_object.self_uri:
        logger.warning(f"Mock bad id {drs_object.self_uri}")
        drs_object.errors = ["Mock error BAD_ID"]
        return None

    with open(Path(os.getcwd(), f"{drs_object.name}.golden"), "rb") as f:
        f.seek(start)
        data = f.read(length_)

    (fd, name,) = tempfile.mkstemp(
        prefix=f"{drs_object.name}.{start}.{size}.",
        suffix=".part",
        dir=str(destination_path),
    )
    with os.fdopen(fd, "wb") as fp:
        sleep_duration = random.randint(1, 3)
        await asyncio.sleep(delay=sleep_duration)
        fp.write(data)
        self.statistics.set_max_files_open()
        fp.close()

    return Path(name)

get_object(object_id, verbose=False) async

Fetch the object from repository DRS Service.

See https://ga4gh.github.io/data-repository-service-schemas/preview/release/drs-1.0.0/docs/#_getobject.

Parameters:

Name Type Description Default
object_id str
required

Returns:

Source code in drs_downloader/clients/mock.py 
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
async def get_object(self, object_id: str, verbose: bool = False) -> DrsObject:
    """Fetch the object from repository DRS Service.

    See https://ga4gh.github.io/data-repository-service-schemas/preview/release/drs-1.0.0/docs/#_getobject.

    Args:
        object_id:

    Returns:

    """
    # """Actually fetch the object.
    #
    # """
    fp = tempfile.TemporaryFile()
    sleep_duration = random.randint(1, 3)
    await asyncio.sleep(delay=sleep_duration)
    fp.write(b"get object")
    self.statistics.set_max_files_open()
    fp.close()

    id_ = str(uuid.uuid4())
    name_ = f"file-{id_}.txt"

    line = b"Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n"  # noqa
    line_len = len(line)
    number_of_lines = int(random.randint(line_len, MAX_SIZE_OF_OBJECT) / line_len)
    lines = line * number_of_lines
    size_ = len(lines)

    # write it for testing
    destination_dir = Path(os.getcwd())
    destination_dir.mkdir(parents=True, exist_ok=True)
    with open(Path(f"{destination_dir}/{name_}.golden"), "wb") as f:
        f.write(lines)

    checksum = Checksum(hashlib.new("md5", lines).hexdigest(), type="md5")

    # simulate an incorrect MD5
    if object_id == BAD_MD5:
        checksum = Checksum(hashlib.new("md5", line).hexdigest(), type="md5")

    # simulate an incorrect size
    if object_id == INCORRECT_SIZE:
        size_ += 1000

    return DrsObject(
        self_uri=object_id,
        size=size_,
        # md5, etag, crc32c, trunc512, or sha1
        checksums=[checksum],
        id=id_,
        name=name_,
    )

sign_url(drs_object, verbose=False) async

Simulate url signing by waiting 1-3 seconds, return populated DrsObject

Parameters:

Name Type Description Default
drs_object DrsObject
required

Returns:

Type Description
Optional[DrsObject]

populated DrsObject
Source code in drs_downloader/clients/mock.py 
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
async def sign_url(self, drs_object: DrsObject, verbose: bool = False) -> Optional[DrsObject]:
    """Simulate url signing by waiting 1-3 seconds, return populated DrsObject

    Args:
        drs_object:

    Returns:
        populated DrsObject
    """
    # simulate a failed signature
    if drs_object.id == BAD_SIGNATURE:
        return None

    # here we sleep while the file is open and measure total files open
    fp = tempfile.TemporaryFile()
    sleep_duration = random.randint(1, 3)
    await asyncio.sleep(delay=sleep_duration)
    fp.write(b"sign url")
    self.statistics.set_max_files_open()
    fp.close()

    # provide expected result, e.g. X-Signature
    access_url = f"{drs_object.self_uri}?X-Signature={uuid.uuid1()}"
    # place it in the right spot in the drs object
    drs_object.access_methods.append(AccessMethod(access_url=access_url, type="gs"))

    return drs_object

manifest_all_ok(number_of_object_ids)

Generate a test manifest, a tsv file with valid drs identifiers.

Source code in drs_downloader/clients/mock.py 
159
160
161
162
163
164
165
166
167
def manifest_all_ok(number_of_object_ids):
    """Generate a test manifest, a tsv file with valid drs identifiers."""
    ids_from_manifest = [str(uuid.uuid4()) for _ in range(number_of_object_ids)]
    tsv_file = tempfile.NamedTemporaryFile(delete=False, mode="w")
    tsv_file.write("ga4gh_drs_uri\n")
    for id_ in ids_from_manifest:
        tsv_file.write(f"drs://{id_}\n")
    tsv_file.close()
    return tsv_file

manifest_bad_file_size()

Generate a test manifest, a tsv file with 2 valid drs identifiers and one that will create an incorrect file.

Source code in drs_downloader/clients/mock.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
def manifest_bad_file_size():
    """Generate a test manifest, a tsv file with 2 valid drs identifiers and one that will create an incorrect file."""
    ids_from_manifest = [
        "drs://" + str(uuid.uuid4()),
        INCORRECT_SIZE,
        "drs://" + str(uuid.uuid4()),
    ]
    tsv_file = tempfile.NamedTemporaryFile(delete=False, mode="w")
    tsv_file.write("ga4gh_drs_uri\n")
    for id_ in ids_from_manifest:
        tsv_file.write(f"{id_}\n")
    tsv_file.close()
    return tsv_file

manifest_bad_id_for_download()

Generate a test manifest, a tsv file with 2 valid drs identifiers and one that will create an incorrect file.

Source code in drs_downloader/clients/mock.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def manifest_bad_id_for_download():
    """Generate a test manifest, a tsv file with 2 valid drs identifiers and one that will create an incorrect file."""
    ids_from_manifest = [
        "drs://" + str(uuid.uuid4()),
        "drs://" + str(uuid.uuid4()),
        BAD_ID,
        "drs://" + str(uuid.uuid4()),
    ]
    tsv_file = tempfile.NamedTemporaryFile(delete=False, mode="w")
    tsv_file.write("ga4gh_drs_uri\n")
    for id_ in ids_from_manifest:
        tsv_file.write(f"{id_}\n")
    tsv_file.close()
    return tsv_file