Coverage for tarjinja/tar.py: 76%

1import io 

2import os 

3import tarfile 

4import time 

5from collections.abc import Generator 

6from logging import getLogger 

7 

8from .iface import Input, Output 

9 

10log = getLogger(__name__) 

11 

12 

13class TarInput(Input): 

14 def __init__(self, ifn: str, **kwargs): 

15 super().__init__(ifn) 

16 self.tf = tarfile.open(ifn) 

17 self.encoding = "utf-8" 

18 

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

20 # does not work? 

21 while True: 

22 n = self.tf.next() 

23 log.info("next %s", n) 

24 if n is None: 

25 break 

26 if not n.isfile(): 

27 continue 

28 yield n.name, n.mode, n.mtime 

29 

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

31 for n in self.tf.getmembers(): 

32 if not n.isfile(): 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 continue 

34 yield n.name, n.mode, n.mtime 

35 

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

37 return self.tf.extractfile(fn).read().decode(self.encoding) 

38 

39 

40class TarOutput(Output): 

41 def __init__(self, ofn: str, **kwargs): 

42 super().__init__(ofn) 

43 _base, ext = os.path.splitext(ofn) 

44 ext = ext[1:] 

45 if ext == "tar": 

46 ext = "" 

47 self.tf = tarfile.open(ofn, f"w:{ext}") 

48 self.encoding = "utf-8" 

49 self.owner = "root" 

50 self.group = "root" 

51 

52 def writefile(self, fn: str, content: str, mode: int, ts: float | None = None): 

53 bcont = content.encode(self.encoding) 

54 tinfo = tarfile.TarInfo(fn) 

55 tinfo.size = len(bcont) 

56 tinfo.mode = mode 

57 tinfo.type = tarfile.REGTYPE 

58 tinfo.uname = self.owner 

59 tinfo.gname = self.group 

60 if ts is None: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true

61 tinfo.mtime = time.time() 

62 else: 

63 tinfo.mtime = ts 

64 self.tf.addfile(tinfo, io.BytesIO(bcont)) 

65 

66 def finish(self): 

67 self.tf.close()