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 

6 

7import requests 

8from sqlalchemy.orm import object_session 

9 

10from postrfp import conf 

11from postrfp.mail.postmark import api_headers 

12from postrfp.model import QuestionInstance, QElement, Issue 

13from postrfp.model.questionnaire.answering import Answer 

14from .constants import API_VERSION_FILEPATH 

15 

16 

17def api_version_resource() -> Traversable: 

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

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

20 

21 

22def read_api_version() -> str: 

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

24 return api_version_resource().read_text().strip() 

25 

26 

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

28 session = object_session(issue) 

29 assert session is not None 

30 issue.answers.delete() 

31 

32 q = ( 

33 session.query(QElement, QuestionInstance) 

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

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

36 ) 

37 

38 if factory is None: 

39 

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

41 return "Autoanswer" 

42 

43 for q_el, q_instance in q: 

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

45 a.answer = factory(a) 

46 session.add(a) 

47 

48 

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

50 from postrfp.model.audit import evt_types 

51 

52 headers = api_headers(key=api_key) 

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

54 resp = requests.get( 

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

56 headers=headers, 

57 params=payload, 

58 timeout=5, 

59 ) 

60 

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

62 if save_disc: 

63 save_templates_to_disc(api_key, template_labels) 

64 

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

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

67 return ev_set - tmpl_by_alias 

68 

69 

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

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

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

73 tmpl_dir = root_dir / _dir 

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

75 

76 headers = api_headers(key=api_key) 

77 for t in template_labels: 

78 alias = t["Alias"] 

79 tmpl_response = requests.get( 

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

81 ) 

82 tmpl_response.raise_for_status() 

83 tmpl = tmpl_response.json() 

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

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

86 

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

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

89 

90 

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

92 import pymustache # type: ignore 

93 

94 headers = api_headers(key=api_key) 

95 

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

97 

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

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

100 tmpl_alias = json_path.stem 

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

102 pymustache.compiled(tmpl_txt) # validates the template 

103 tmpl_json["TextBody"] = tmpl_txt 

104 

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

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

107 if not response.ok: 

108 response.raise_for_status()