Coverage for tarjinja/git.py: 30%

36 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-09-05 14:24 +0000

1import io 

2import os 

3import tempfile 

4import time 

5from collections.abc import Generator 

6from logging import getLogger 

7 

8import git 

9 

10from .iface import Input 

11 

12log = getLogger(__name__) 

13 

14 

15class GitInput(Input): 

16 def __init__( 

17 self, ifn: str, name: str, branch: str = "master", tmpdir: str | None = None 

18 ): 

19 log.debug("ifn=%s, name=%s, branch=%s, tmpdir=%s", ifn, name, branch, tmpdir) 

20 super().__init__(ifn) 

21 if tmpdir is None: 

22 self.tmpd = tempfile.TemporaryDirectory() 

23 self.tmpdir = self.tmpd.name 

24 else: 

25 self.tmpdir = tmpdir 

26 dirn = os.path.join(self.tmpdir, name) 

27 if os.path.isdir(dirn): 

28 self.repo = git.repo.Repo(dirn) 

29 self.repo.remote().fetch(refspec=branch, depth=1) 

30 else: 

31 self.repo = git.repo.Repo.clone_from( 

32 self.ifn, dirn, bare=True, branch=branch, depth=1 

33 ) 

34 self.ts = time.time() 

35 self.files = {} 

36 

37 def walk(self) -> Generator[tuple[str, int, float], None, None]: 

38 for x in self.repo.tree().traverse(): 

39 if x.type != "blob": 

40 continue 

41 self.files[x.path] = x 

42 yield x.path, x.mode, self.ts 

43 

44 def readfile(self, fn: str) -> str: 

45 try: 

46 return io.TextIOWrapper( 

47 io.BytesIO(self.files.get(fn).data_stream.read()) 

48 ).read() 

49 except UnicodeDecodeError as e: 

50 log.warning("cannot decode: %s", e) 

51 return ""