Coverage for rfpy/tools.py: 100%

1import os 

2import json 

3from pathlib import Path 

4from typing import Optional 

5 

6import requests 

7from sqlalchemy.orm import object_session, Session 

8 

9from rfpy import conf 

10from rfpy.model import QuestionInstance, QElement, Answer 

11from rfpy.mail.postmark import api_headers 

12 

13""" 

14TEST_SESSION is a global variable used for transaction hacks in functional tests 

15""" 

16 

17TEST_SESSION: Optional[Session] = None 

18 

19 

20def populate_answers(issue, factory=None): 

21 session = object_session(issue) 

22 

23 issue.answers.delete() 

24 

25 q = ( 

26 session.query(QElement, QuestionInstance) 

27 .filter(QElement.question_id == QuestionInstance.question_def_id) 

28 .filter(QuestionInstance.project == issue.project) 

29 ) 

30 

31 if factory is None: 

32 

33 def factory(answer): 

34 return "Autoanswer" 

35 

36 for q_el, q_instance in q: 

37 a = Answer(issue=issue, element=q_el, question_instance=q_instance) 

38 a.answer = factory(a) 

39 session.add(a) 

40 

41 

42def banner_files(tmpl, dirpath, dry_run=True): # pragma: no cover 

43 """ 

44 Add a banner to the top of a bunch of filenames 

45 

46 e.g. banner_files('<!-- nunjucks %s -->\n\n', 'src/js/views/') 

47 """ 

48 for dirpath, _dirnames, filenames in os.walk(dirpath): 

49 for fname in filenames: 

50 fpath = os.path.join(dirpath, fname) 

51 if dry_run: 

52 print("Dry Run: prepending %s to %s" % (tmpl % fpath, fpath)) 

53 else: 

54 with open(fpath, "r+") as f: 

55 content = f.read() 

56 f.seek(0, 0) 

57 f.write(tmpl % fpath) 

58 f.write(content) 

59 

60 

61def list_missing_templates(api_key, save_disc=True): 

62 from rfpy.model.audit import evt_types 

63 

64 headers = api_headers(key=api_key) 

65 payload = {"Count": 100, "Offset": 0} 

66 resp = requests.get( 

67 "https://api.postmarkapp.com/templates", headers=headers, params=payload 

68 ) 

69 

70 template_labels = resp.json()["Templates"] 

71 if save_disc: 

72 save_templates_to_disc(api_key, template_labels) 

73 

74 tmpl_by_alias = {t["Alias"] for t in template_labels} 

75 ev_set = {e for e in dir(evt_types) if not e.startswith("__")} 

76 return ev_set - tmpl_by_alias 

77 

78 

79def save_templates_to_disc(api_key, template_labels): 

80 root_dir = Path("/tmp/postmarktmpls/") 

81 for _dir in ("json", "txt"): 

82 tmpl_dir = root_dir / _dir 

83 tmpl_dir.mkdir(parents=True, exist_ok=True) 

84 

85 headers = api_headers(key=api_key) 

86 for t in template_labels: 

87 alias = t["Alias"] 

88 tmpl_response = requests.get( 

89 f"https://api.postmarkapp.com/templates/{alias}", headers=headers 

90 ) 

91 tmpl_response.raise_for_status() 

92 tmpl = tmpl_response.json() 

93 with open("/tmp/postmarktmpls/json/%s.json" % alias, "w") as f: 

94 json.dump(tmpl, f, indent=2) 

95 

96 with open("/tmp/postmarktmpls/txt/%s.txt" % alias, "w") as ftxt: 

97 print(tmpl["TextBody"], file=ftxt) 

98 

99 

100def upload_templates_to_postmark(api_key): 

101 import pymustache # type: ignore 

102 

103 headers = api_headers(key=api_key) 

104 

105 json_dir = Path(conf.CONF.template_dir).parent / Path("json") 

106 

107 for json_path in json_dir.glob("*.json"): 

108 tmpl_json = json.loads(json_path.read_text()) 

109 tmpl_alias = json_path.stem 

110 tmpl_txt = (json_dir / f"../txt/{tmpl_alias}.txt").read_text() 

111 pymustache.compiled(tmpl_txt) # validates the template 

112 tmpl_json["TextBody"] = tmpl_txt 

113 

114 path = f"https://api.postmarkapp.com/templates/{tmpl_alias}" 

115 response = requests.put(path, headers=headers, json=tmpl_json) 

116 if not response.ok: 

117 response.raise_for_status()