1# Copyright (C) 2010, 2011 Sebastian Thiel (byronimo@gmail.com) and contributors
2#
3# This module is part of GitDB and is released under
4# the New BSD License: http://www.opensource.org/licenses/bsd-license.php
5"""Base classes for object db testing"""
6from gitdb.test.lib import (
7    with_rw_directory,
8    with_packs_rw,
9    fixture_path,
10    TestBase
11)
12
13from gitdb.stream import (
14    Sha1Writer,
15    ZippedStoreShaWriter
16)
17
18from gitdb.base import (
19    IStream,
20    OStream,
21    OInfo
22)
23
24from gitdb.exc import BadObject
25from gitdb.typ import str_blob_type
26
27from io import BytesIO
28
29from struct import pack
30
31
32__all__ = ('TestDBBase', 'with_rw_directory', 'with_packs_rw', 'fixture_path')
33
34
35class TestDBBase(TestBase):
36
37    """Base class providing testing routines on databases"""
38
39    # data
40    two_lines = b'1234\nhello world'
41    all_data = (two_lines, )
42
43    def _assert_object_writing_simple(self, db):
44        # write a bunch of objects and query their streams and info
45        null_objs = db.size()
46        ni = 250
47        for i in range(ni):
48            data = pack(">L", i)
49            istream = IStream(str_blob_type, len(data), BytesIO(data))
50            new_istream = db.store(istream)
51            assert new_istream is istream
52            assert db.has_object(istream.binsha)
53
54            info = db.info(istream.binsha)
55            assert isinstance(info, OInfo)
56            assert info.type == istream.type and info.size == istream.size
57
58            stream = db.stream(istream.binsha)
59            assert isinstance(stream, OStream)
60            assert stream.binsha == info.binsha and stream.type == info.type
61            assert stream.read() == data
62        # END for each item
63
64        assert db.size() == null_objs + ni
65        shas = list(db.sha_iter())
66        assert len(shas) == db.size()
67        assert len(shas[0]) == 20
68
69    def _assert_object_writing(self, db):
70        """General tests to verify object writing, compatible to ObjectDBW
71        **Note:** requires write access to the database"""
72        # start in 'dry-run' mode, using a simple sha1 writer
73        ostreams = (ZippedStoreShaWriter, None)
74        for ostreamcls in ostreams:
75            for data in self.all_data:
76                dry_run = ostreamcls is not None
77                ostream = None
78                if ostreamcls is not None:
79                    ostream = ostreamcls()
80                    assert isinstance(ostream, Sha1Writer)
81                # END create ostream
82
83                prev_ostream = db.set_ostream(ostream)
84                assert type(prev_ostream) in ostreams or prev_ostream in ostreams
85                istream = IStream(str_blob_type, len(data), BytesIO(data))
86
87                # store returns same istream instance, with new sha set
88                my_istream = db.store(istream)
89                sha = istream.binsha
90                assert my_istream is istream
91                assert db.has_object(sha) != dry_run
92                assert len(sha) == 20
93
94                # verify data - the slow way, we want to run code
95                if not dry_run:
96                    info = db.info(sha)
97                    assert str_blob_type == info.type
98                    assert info.size == len(data)
99
100                    ostream = db.stream(sha)
101                    assert ostream.read() == data
102                    assert ostream.type == str_blob_type
103                    assert ostream.size == len(data)
104                else:
105                    self.assertRaises(BadObject, db.info, sha)
106                    self.assertRaises(BadObject, db.stream, sha)
107
108                    # DIRECT STREAM COPY
109                    # our data hase been written in object format to the StringIO
110                    # we pasesd as output stream. No physical database representation
111                    # was created.
112                    # Test direct stream copy of object streams, the result must be
113                    # identical to what we fed in
114                    ostream.seek(0)
115                    istream.stream = ostream
116                    assert istream.binsha is not None
117                    prev_sha = istream.binsha
118
119                    db.set_ostream(ZippedStoreShaWriter())
120                    db.store(istream)
121                    assert istream.binsha == prev_sha
122                    new_ostream = db.ostream()
123
124                    # note: only works as long our store write uses the same compression
125                    # level, which is zip_best
126                    assert ostream.getvalue() == new_ostream.getvalue()
127            # END for each data set
128        # END for each dry_run mode
129