11import codecs
22from io import BytesIO
3+ from typing import Callable , Generator
34
45import pytest
56
67import pygit2
8+ from pygit2 import Blob , Filter , FilterSource , Repository
79from pygit2 .enums import BlobFilter
810from pygit2 .errors import Passthrough
911
1012
11- def _rot13 (data ) :
13+ def _rot13 (data : bytes ) -> bytes :
1214 return codecs .encode (data .decode ('utf-8' ), 'rot_13' ).encode ('utf-8' )
1315
1416
1517class _Rot13Filter (pygit2 .Filter ):
1618 attributes = 'text'
1719
18- def write (self , data , src , write_next ):
20+ def write (
21+ self ,
22+ data : bytes ,
23+ src : FilterSource ,
24+ write_next : Callable [[bytes ], None ],
25+ ) -> None :
1926 return super ().write (_rot13 (data ), src , write_next )
2027
2128
2229class _BufferedFilter (pygit2 .Filter ):
2330 attributes = 'text'
2431
25- def __init__ (self ):
32+ def __init__ (self ) -> None :
2633 super ().__init__ ()
2734 self .buf = BytesIO ()
2835
29- def write (self , data , src , write_next ):
36+ def write (
37+ self ,
38+ data : bytes ,
39+ src : FilterSource ,
40+ write_next : Callable [[bytes ], None ],
41+ ) -> None :
3042 self .buf .write (data )
3143
32- def close (self , write_next ) :
44+ def close (self , write_next : Callable [[ bytes ], None ]) -> None :
3345 write_next (_rot13 (self .buf .getvalue ()))
3446
3547
3648class _PassthroughFilter (_Rot13Filter ):
37- def check (self , src , attr_values ) :
49+ def check (self , src : FilterSource , attr_values : list [ str | None ]) -> None :
3850 assert attr_values == [None ]
3951 assert src .repo
4052 raise Passthrough
@@ -45,36 +57,37 @@ class _UnmatchedFilter(_Rot13Filter):
4557
4658
4759@pytest .fixture
48- def rot13_filter ():
60+ def rot13_filter () -> Generator [ None , None , None ] :
4961 pygit2 .filter_register ('rot13' , _Rot13Filter )
5062 yield
5163 pygit2 .filter_unregister ('rot13' )
5264
5365
5466@pytest .fixture
55- def passthrough_filter ():
67+ def passthrough_filter () -> Generator [ None , None , None ] :
5668 pygit2 .filter_register ('passthrough-rot13' , _PassthroughFilter )
5769 yield
5870 pygit2 .filter_unregister ('passthrough-rot13' )
5971
6072
6173@pytest .fixture
62- def buffered_filter ():
74+ def buffered_filter () -> Generator [ None , None , None ] :
6375 pygit2 .filter_register ('buffered-rot13' , _BufferedFilter )
6476 yield
6577 pygit2 .filter_unregister ('buffered-rot13' )
6678
6779
6880@pytest .fixture
69- def unmatched_filter ():
81+ def unmatched_filter () -> Generator [ None , None , None ] :
7082 pygit2 .filter_register ('unmatched-rot13' , _UnmatchedFilter )
7183 yield
7284 pygit2 .filter_unregister ('unmatched-rot13' )
7385
7486
75- def test_filter (testrepo , rot13_filter ) :
87+ def test_filter (testrepo : Repository , rot13_filter : Filter ) -> None :
7688 blob_oid = testrepo .create_blob_fromworkdir ('bye.txt' )
7789 blob = testrepo [blob_oid ]
90+ assert isinstance (blob , Blob )
7891 flags = BlobFilter .CHECK_FOR_BINARY | BlobFilter .ATTRIBUTES_FROM_HEAD
7992 assert b'olr jbeyq\n ' == blob .data
8093 with pygit2 .BlobIO (blob ) as reader :
@@ -83,9 +96,10 @@ def test_filter(testrepo, rot13_filter):
8396 assert b'bye world\n ' == reader .read ()
8497
8598
86- def test_filter_buffered (testrepo , buffered_filter ) :
99+ def test_filter_buffered (testrepo : Repository , buffered_filter : Filter ) -> None :
87100 blob_oid = testrepo .create_blob_fromworkdir ('bye.txt' )
88101 blob = testrepo [blob_oid ]
102+ assert isinstance (blob , Blob )
89103 flags = BlobFilter .CHECK_FOR_BINARY | BlobFilter .ATTRIBUTES_FROM_HEAD
90104 assert b'olr jbeyq\n ' == blob .data
91105 with pygit2 .BlobIO (blob ) as reader :
@@ -94,9 +108,10 @@ def test_filter_buffered(testrepo, buffered_filter):
94108 assert b'bye world\n ' == reader .read ()
95109
96110
97- def test_filter_passthrough (testrepo , passthrough_filter ) :
111+ def test_filter_passthrough (testrepo : Repository , passthrough_filter : Filter ) -> None :
98112 blob_oid = testrepo .create_blob_fromworkdir ('bye.txt' )
99113 blob = testrepo [blob_oid ]
114+ assert isinstance (blob , Blob )
100115 flags = BlobFilter .CHECK_FOR_BINARY | BlobFilter .ATTRIBUTES_FROM_HEAD
101116 assert b'bye world\n ' == blob .data
102117 with pygit2 .BlobIO (blob ) as reader :
@@ -105,9 +120,10 @@ def test_filter_passthrough(testrepo, passthrough_filter):
105120 assert b'bye world\n ' == reader .read ()
106121
107122
108- def test_filter_unmatched (testrepo , unmatched_filter ) :
123+ def test_filter_unmatched (testrepo : Repository , unmatched_filter : Filter ) -> None :
109124 blob_oid = testrepo .create_blob_fromworkdir ('bye.txt' )
110125 blob = testrepo [blob_oid ]
126+ assert isinstance (blob , Blob )
111127 flags = BlobFilter .CHECK_FOR_BINARY | BlobFilter .ATTRIBUTES_FROM_HEAD
112128 assert b'bye world\n ' == blob .data
113129 with pygit2 .BlobIO (blob ) as reader :
@@ -116,7 +132,7 @@ def test_filter_unmatched(testrepo, unmatched_filter):
116132 assert b'bye world\n ' == reader .read ()
117133
118134
119- def test_filter_cleanup (dirtyrepo , rot13_filter ) :
135+ def test_filter_cleanup (dirtyrepo : Repository , rot13_filter : Filter ) -> None :
120136 # Indirectly test that pygit2_filter_cleanup has the GIL
121137 # before calling pygit2_filter_payload_free.
122138 dirtyrepo .diff ()
0 commit comments