Coverage for src/lektor_ng/buildfailures.py: 100%

1import errno 

2import hashlib 

3import json 

4import os 

5from traceback import TracebackException 

6 

7 

8class BuildFailure: 

9 def __init__(self, data): 

10 self.data = data 

11 

12 @classmethod 

13 def from_exc_info(cls, artifact_name, exc_info): 

14 te = TracebackException(*exc_info) 

15 # NB: we have dropped werkzeug's support for Paste's __traceback_hide__ 

16 # frame local. 

17 return cls( 

18 { 

19 "artifact": artifact_name, 

20 "exception": "".join(te.format_exception_only()).strip(), 

21 "traceback": "".join(te.format()).strip(), 

22 } 

23 ) 

24 

25 def to_json(self): 

26 return self.data 

27 

28 

29class FailureController: 

30 def __init__(self, pad, destination_path): 

31 self.pad = pad 

32 self.path = os.path.join( 

33 os.path.abspath(os.path.join(pad.db.env.root_path, destination_path)), 

34 ".lektor", 

35 "failures", 

36 ) 

37 

38 def get_filename(self, artifact_name): 

39 return os.path.join(self.path, hashlib.md5(artifact_name.encode("utf-8")).hexdigest()) + ".json" 

40 

41 def lookup_failure(self, artifact_name): 

42 """Looks up a failure for the given artifact name.""" 

43 fn = self.get_filename(artifact_name) 

44 try: 

45 with open(fn, encoding="utf-8") as f: 

46 return BuildFailure(json.load(f)) 

47 except OSError as e: 

48 if e.errno != errno.ENOENT: 

49 raise 

50 return None 

51 

52 def clear_failure(self, artifact_name): 

53 """Clears a stored failure.""" 

54 try: 

55 os.unlink(self.get_filename(artifact_name)) 

56 except OSError as e: 

57 if e.errno != errno.ENOENT: 

58 raise 

59 

60 def store_failure(self, artifact_name, exc_info): 

61 """Stores a failure from an exception info tuple.""" 

62 fn = self.get_filename(artifact_name) 

63 try: 

64 os.makedirs(os.path.dirname(fn)) 

65 except OSError: 

66 pass 

67 with open(fn, mode="w", encoding="utf-8") as f: 

68 json.dump(BuildFailure.from_exc_info(artifact_name, exc_info).to_json(), f) 

69 f.write("\n")