File: hashed_file.py

package info (click to toggle)
game-data-packager 87
  • links: PTS, VCS
  • area: contrib
  • in suites: sid
  • size: 33,392 kB
  • sloc: python: 15,387; sh: 704; ansic: 95; makefile: 50
file content (147 lines) | stat: -rwxr-xr-x 4,765 bytes parent folder | download | duplicates (3)
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
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
#!/usr/bin/python3
# encoding=utf-8
#
# Copyright © 2016 Simon McVittie <smcv@debian.org>
# SPDX-License-Identifier: GPL-2.0-or-later

import hashlib
import os
import sys
import unittest
from typing import (BinaryIO, cast)

if 'GDP_UNINSTALLED' not in os.environ:
    sys.path.insert(0, '/usr/share/game-data-packager')
    sys.path.insert(0, '/usr/share/games/game-data-packager')

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from game_data_packager.command_line import (TerminalProgress)  # noqa: E402
from game_data_packager.data import (HashedFile)  # noqa: E402


class ZeroReader:
    def __init__(self, total: int) -> None:
        self.done = 0
        self.total = total

    def read(self, max_bytes: int) -> bytes:
        ret = min(max_bytes, self.total - self.done)
        self.total -= ret
        return b'\x00' * ret


SIZE = 30 * 1024 * 1024


class HashedFileTestCase(unittest.TestCase):
    def setUp(self) -> None:
        hasher = hashlib.new('md5')
        hasher.update(b'hello')
        self.HELLO_MD5 = hasher.hexdigest()
        hasher.update(b', world!')
        self.HELLO_WORLD_MD5 = hasher.hexdigest()

        hasher = hashlib.new('sha1')
        hasher.update(b'hello')
        self.HELLO_SHA1 = hasher.hexdigest()
        hasher.update(b', world!')
        self.HELLO_WORLD_SHA1 = hasher.hexdigest()

        hasher = hashlib.new('sha256')
        hasher.update(b'hello')
        self.HELLO_SHA256 = hasher.hexdigest()
        hasher.update(b', world!')
        self.HELLO_WORLD_SHA256 = hasher.hexdigest()

    def test_attrs(self) -> None:
        hf = HashedFile('hello.txt')
        self.assertIs(hf.have_hashes, False)

        hf.md5 = self.HELLO_MD5
        self.assertIs(hf.have_hashes, True)
        self.assertEqual(hf.md5, self.HELLO_MD5)

        with self.assertRaises(AssertionError):
            hf.md5 = None

        hf.sha1 = self.HELLO_SHA1
        self.assertIs(hf.have_hashes, True)
        self.assertEqual(hf.sha1, self.HELLO_SHA1)

        hf.sha256 = self.HELLO_SHA256
        self.assertIs(hf.have_hashes, True)
        self.assertEqual(hf.sha256, self.HELLO_SHA256)

    def test_matches(self) -> None:
        first = HashedFile('hello.txt')
        first.md5 = self.HELLO_MD5

        second = HashedFile('hello.txt')
        second.sha1 = self.HELLO_SHA1

        with self.assertRaises(ValueError):
            first.matches(second)

        with self.assertRaises(ValueError):
            second.matches(first)

        second.md5 = self.HELLO_MD5
        self.assertIs(first.matches(second), True)
        self.assertIs(second.matches(first), True)

        second = HashedFile('hello_world.txt')
        second.md5 = self.HELLO_WORLD_MD5
        self.assertIs(first.matches(second), False)
        self.assertIs(second.matches(first), False)

    def test_progress(self) -> None:
        print('', file=sys.stderr)
        HashedFile.from_file('progress.bin',
                             cast(BinaryIO, ZeroReader(SIZE)),
                             size=SIZE,
                             progress=TerminalProgress(interval=0.1))
        print('', file=sys.stderr)
        HashedFile.from_file('progress.bin',
                             cast(BinaryIO, ZeroReader(SIZE)),
                             progress=TerminalProgress(interval=0.1))
        print('', file=sys.stderr)

        HashedFile.from_file('progress.bin',
                             cast(BinaryIO, ZeroReader(SIZE)),
                             size=SIZE)
        HashedFile.from_file('progress.bin',
                             cast(BinaryIO, ZeroReader(SIZE)))

        print('', file=sys.stderr)
        HashedFile.from_concatenated_files(
                'concatenated.bin',
                [cast(BinaryIO, ZeroReader(SIZE)),
                 cast(BinaryIO, ZeroReader(SIZE))],
                size=2 * SIZE,
                progress=TerminalProgress(interval=0.1))
        print('', file=sys.stderr)
        HashedFile.from_concatenated_files(
                'concatenated.bin',
                [cast(BinaryIO, ZeroReader(SIZE)),
                 cast(BinaryIO, ZeroReader(SIZE))],
                progress=TerminalProgress(interval=0.1))
        print('', file=sys.stderr)

        HashedFile.from_concatenated_files(
                'concatenated.bin',
                [cast(BinaryIO, ZeroReader(SIZE)),
                 cast(BinaryIO, ZeroReader(SIZE))],
                size=2 * SIZE)
        HashedFile.from_concatenated_files(
                'concatenated.bin',
                [cast(BinaryIO, ZeroReader(SIZE)),
                 cast(BinaryIO, ZeroReader(SIZE))])

    def tearDown(self) -> None:
        pass


if __name__ == '__main__':
    from gdp_test_common import main
    main()