Coverage for postrfp / shared / tools.py: 100%

1import json 

2import importlib.resources as resources 

3from importlib.resources.abc import Traversable 

4from pathlib import Path 

5from typing import Callable, TYPE_CHECKING 

6 

7import requests 

8 

9 

10from postrfp import conf 

11from postrfp.mail.postmark import api_headers 

12 

13from .constants import API_VERSION_FILEPATH 

14 

15if TYPE_CHECKING: 

16 from postrfp.model import Issue 

17 

18 

19def api_version_resource() -> Traversable: 

20 package_name, filename = tuple(API_VERSION_FILEPATH.rsplit(".", 1)) 

21 return resources.files(package_name).joinpath(filename) 

22 

23 

24def read_api_version() -> str: 

25 """Return the API version string stored in the package.""" 

26 return api_version_resource().read_text().strip() 

27 

28 

29def populate_answers(issue: "Issue", factory: Callable | None = None) -> None: 

30 from sqlalchemy.orm import object_session 

31 

32 from postrfp.model import QuestionInstance, QElement 

33 from postrfp.model.questionnaire.answering import Answer 

34 

35 session = object_session(issue) 

36 assert session is not None 

37 issue.answers.delete() 

38 

39 q = ( 

40 session.query(QElement, QuestionInstance) 

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

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

43 ) 

44 

45 if factory is None: 

46 

47 def factory(answer: Answer) -> str: 

48 return "Autoanswer" 

49 

50 for q_el, q_instance in q: 

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

52 a.answer = factory(a) 

53 session.add(a) 

54 

55 

56def list_missing_templates(api_key: str, save_disc: bool = True) -> set[str]: 

57 from postrfp.model.audit import evt_types 

58 

59 headers = api_headers(key=api_key) 

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

61 resp = requests.get( 

62 "https://api.postmarkapp.com/templates", 

63 headers=headers, 

64 params=payload, 

65 timeout=5, 

66 ) 

67 

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

69 if save_disc: 

70 save_templates_to_disc(api_key, template_labels) 

71 

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

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

74 return ev_set - tmpl_by_alias 

75 

76 

77def save_templates_to_disc(api_key: str, template_labels: list[dict]) -> None: 

78 root_dir = Path("/tmp/postmarktmpls/") # nosec 

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

80 tmpl_dir = root_dir / _dir 

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

82 

83 headers = api_headers(key=api_key) 

84 for t in template_labels: 

85 alias = t["Alias"] 

86 tmpl_response = requests.get( 

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

88 ) 

89 tmpl_response.raise_for_status() 

90 tmpl = tmpl_response.json() 

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

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

93 

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

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

96 

97 

98def upload_templates_to_postmark(api_key: str) -> None: # pragma: no cover 

99 import pymustache # type: ignore 

100 

101 headers = api_headers(key=api_key) 

102 

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

104 

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

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

107 tmpl_alias = json_path.stem 

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

109 pymustache.compiled(tmpl_txt) # validates the template 

110 tmpl_json["TextBody"] = tmpl_txt 

111 

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

113 response = requests.put(path, headers=headers, json=tmpl_json, timeout=5) 

114 if not response.ok: 

115 response.raise_for_status()